fix(log-query): panic on prometheus (#5429)

* fix(log-query): panic on prometheus

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* fix test environment setup

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

---------

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

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

@@ -54,7 +54,7 @@ runs:
         PROFILE_TARGET: ${{ inputs.cargo-profile == 'dev' && 'debug' || inputs.cargo-profile }}
       with:
         artifacts-dir: ${{ inputs.artifacts-dir }}
-        target-file: ./target/$PROFILE_TARGET/greptime
+        target-files: ./target/$PROFILE_TARGET/greptime
         version: ${{ inputs.version }}
         working-dir: ${{ inputs.working-dir }}
 
@@ -72,6 +72,6 @@ runs:
       if: ${{ inputs.build-android-artifacts == 'true' }}
       with:
         artifacts-dir: ${{ inputs.artifacts-dir }}
-        target-file: ./target/aarch64-linux-android/release/greptime
+        target-files: ./target/aarch64-linux-android/release/greptime
         version: ${{ inputs.version }}
         working-dir: ${{ inputs.working-dir }}

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

@@ -41,8 +41,8 @@ runs:
         image-name: ${{ inputs.image-name }}
         image-tag: ${{ inputs.version }}
         docker-file: docker/ci/ubuntu/Dockerfile
-        amd64-artifact-name: greptime-linux-amd64-pyo3-${{ inputs.version }}
-        arm64-artifact-name: greptime-linux-arm64-pyo3-${{ inputs.version }}
+        amd64-artifact-name: greptime-linux-amd64-${{ inputs.version }}
+        arm64-artifact-name: greptime-linux-arm64-${{ inputs.version }}
         platforms: linux/amd64,linux/arm64
         push-latest-tag: ${{ inputs.push-latest-tag }}
 

.github/actions/build-linux-artifacts/action.yml

@@ -48,19 +48,7 @@ runs:
         path: /tmp/greptime-*.log
         retention-days: 3
 
-    - name: Build standard greptime
-      uses: ./.github/actions/build-greptime-binary
-      with:
-        base-image: ubuntu
-        features: pyo3_backend,servers/dashboard
-        cargo-profile: ${{ inputs.cargo-profile }}
-        artifacts-dir: greptime-linux-${{ inputs.arch }}-pyo3-${{ inputs.version }}
-        version: ${{ inputs.version }}
-        working-dir: ${{ inputs.working-dir }}
-        image-registry: ${{ inputs.image-registry }}
-        image-namespace: ${{ inputs.image-namespace }}
-
-    - name: Build greptime without pyo3
+    - name: Build greptime
       if: ${{ inputs.dev-mode == 'false' }}
       uses: ./.github/actions/build-greptime-binary
       with:

.github/actions/build-macos-artifacts/action.yml

@@ -90,5 +90,5 @@ runs:
       uses: ./.github/actions/upload-artifacts
       with:
         artifacts-dir: ${{ inputs.artifacts-dir }}
-        target-file: target/${{ inputs.arch }}/${{ inputs.cargo-profile }}/greptime
+        target-files: target/${{ inputs.arch }}/${{ inputs.cargo-profile }}/greptime
         version: ${{ inputs.version }}

.github/actions/build-windows-artifacts/action.yml

@@ -33,15 +33,6 @@ runs:
     - name: Rust Cache
       uses: Swatinem/rust-cache@v2
 
-    - name: Install Python
-      uses: actions/setup-python@v5
-      with:
-        python-version: "3.10"
-
-    - name: Install PyArrow Package
-      shell: pwsh
-      run: pip install pyarrow numpy
-
     - name: Install WSL distribution
       uses: Vampire/setup-wsl@v2
       with:
@@ -76,5 +67,5 @@ runs:
       uses: ./.github/actions/upload-artifacts
       with:
         artifacts-dir: ${{ inputs.artifacts-dir }}
-        target-file: target/${{ inputs.arch }}/${{ inputs.cargo-profile }}/greptime
+        target-files: target/${{ inputs.arch }}/${{ inputs.cargo-profile }}/greptime,target/${{ inputs.arch }}/${{ inputs.cargo-profile }}/greptime.pdb
         version: ${{ inputs.version }}

.github/actions/setup-greptimedb-cluster/with-minio-and-cache.yaml

@@ -5,7 +5,7 @@ meta:
 
     [datanode]
     [datanode.client]
-    timeout = "60s"
+    timeout = "120s"
 datanode:
   configData: |-
     [runtime]
@@ -21,7 +21,7 @@ frontend:
     global_rt_size = 4
 
     [meta_client]
-    ddl_timeout = "60s"
+    ddl_timeout = "120s"
 objectStorage:
   s3:
     bucket: default

.github/actions/setup-greptimedb-cluster/with-minio.yaml

@@ -5,7 +5,7 @@ meta:
     
     [datanode]
     [datanode.client]
-    timeout = "60s"
+    timeout = "120s"
 datanode:
   configData: |-
     [runtime]
@@ -17,7 +17,7 @@ frontend:
     global_rt_size = 4
 
     [meta_client]
-    ddl_timeout = "60s"
+    ddl_timeout = "120s"
 objectStorage:
   s3:
     bucket: default

.github/actions/setup-greptimedb-cluster/with-remote-wal.yaml

@@ -11,7 +11,7 @@ meta:
         
     [datanode]
     [datanode.client]
-    timeout = "60s"
+    timeout = "120s"
 datanode:
   configData: |-
     [runtime]
@@ -28,7 +28,7 @@ frontend:
     global_rt_size = 4
 
     [meta_client]
-    ddl_timeout = "60s"
+    ddl_timeout = "120s"
 objectStorage:
   s3:
     bucket: default

.github/actions/upload-artifacts/action.yml

@@ -4,8 +4,8 @@ inputs:
   artifacts-dir:
     description: Directory to store artifacts
     required: true
-  target-file:
-    description: The path of the target artifact
+  target-files:
+    description: The multiple target files to upload, separated by comma
     required: false
   version:
     description: Version of the artifact
@@ -18,12 +18,16 @@ runs:
   using: composite
   steps:
     - name: Create artifacts directory
-      if: ${{ inputs.target-file != '' }}
+      if: ${{ inputs.target-files != '' }}
       working-directory: ${{ inputs.working-dir }}
       shell: bash
       run: |
-        mkdir -p ${{ inputs.artifacts-dir }} && \
-        cp ${{ inputs.target-file }} ${{ inputs.artifacts-dir }}
+        set -e
+        mkdir -p ${{ inputs.artifacts-dir }}
+        IFS=',' read -ra FILES <<< "${{ inputs.target-files }}"
+        for file in "${FILES[@]}"; do
+          cp "$file" ${{ inputs.artifacts-dir }}/
+        done
 
     # The compressed artifacts will use the following layout:
     # greptime-linux-amd64-pyo3-v0.3.0sha256sum

.github/pull_request_template.md

@@ -4,7 +4,8 @@ I hereby agree to the terms of the [GreptimeDB CLA](https://github.com/GreptimeT
 
 ## What's changed and what's your intention?
 
-__!!! DO NOT LEAVE THIS BLOCK EMPTY !!!__
+<!--    
+ __!!! DO NOT LEAVE THIS BLOCK EMPTY !!!__
 
 Please explain IN DETAIL what the changes are in this PR and why they are needed:
 
@@ -12,9 +13,14 @@ Please explain IN DETAIL what the changes are in this PR and why they are needed
 - How does this PR work? Need a brief introduction for the changed logic (optional)
 - Describe clearly one logical change and avoid lazy messages (optional)
 - Describe any limitations of the current code (optional)
+- Describe if this PR will break **API or data compatibility**  (optional)
+-->
 
-## Checklist
+## PR Checklist
+Please convert it to a draft if some of the following conditions are not met.
 
 - [ ] I have written the necessary rustdoc comments.
 - [ ] I have added the necessary unit tests and integration tests.
 - [ ] This PR requires documentation updates.
+- [ ] API changes are backward compatible.
+- [ ] Schema or data changes are backward compatible.

.github/workflows/dependency-check.yml

@@ -1,9 +1,6 @@
 name: Check Dependencies
 
 on:
-  push:
-    branches:
-      - main
   pull_request:
     branches:
       - main

.github/workflows/dev-build.yml

@@ -29,7 +29,7 @@ on:
       linux_arm64_runner:
         type: choice
         description: The runner uses to build linux-arm64 artifacts
-        default: ec2-c6g.4xlarge-arm64
+        default: ec2-c6g.8xlarge-arm64
         options:
           - ec2-c6g.xlarge-arm64 # 4C8G
           - ec2-c6g.2xlarge-arm64 # 8C16G

.github/workflows/develop.yml

@@ -10,17 +10,6 @@ on:
       - 'docker/**'
       - '.gitignore'
       - 'grafana/**'
-  push:
-    branches:
-      - main
-    paths-ignore:
-      - 'docs/**'
-      - 'config/**'
-      - '**.md'
-      - '.dockerignore'
-      - 'docker/**'
-      - '.gitignore'
-      - 'grafana/**'
   workflow_dispatch:
 
 name: CI
@@ -54,7 +43,7 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        os: [ windows-2022, ubuntu-20.04 ]
+        os: [ ubuntu-20.04 ]
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
@@ -68,6 +57,8 @@ jobs:
           # Shares across multiple jobs
           # Shares with `Clippy` job
           shared-key: "check-lint"
+          cache-all-crates: "true"
+          save-if: ${{ github.ref == 'refs/heads/main' }}
       - name: Run cargo check
         run: cargo check --locked --workspace --all-targets
 
@@ -78,13 +69,8 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - uses: actions-rust-lang/setup-rust-toolchain@v1
-      - name: Rust Cache
-        uses: Swatinem/rust-cache@v2
-        with:
-          # Shares across multiple jobs
-          shared-key: "check-toml"
       - name: Install taplo
-        run: cargo +stable install taplo-cli --version ^0.9 --locked
+        run: cargo +stable install taplo-cli --version ^0.9 --locked --force
       - name: Run taplo
         run: taplo format --check
 
@@ -105,13 +91,15 @@ jobs:
         with:
           # Shares across multiple jobs
           shared-key: "build-binaries"
+          cache-all-crates: "true"
+          save-if: ${{ github.ref == 'refs/heads/main' }}
       - name: Install cargo-gc-bin
         shell: bash
-        run: cargo install cargo-gc-bin
+        run: cargo install cargo-gc-bin --force
       - name: Build greptime binaries
         shell: bash
         # `cargo gc` will invoke `cargo build` with specified args
-        run: cargo gc -- --bin greptime --bin sqlness-runner
+        run: cargo gc -- --bin greptime --bin sqlness-runner --features pg_kvbackend
       - name: Pack greptime binaries
         shell: bash
         run: |
@@ -153,17 +141,12 @@ jobs:
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
       - uses: actions-rust-lang/setup-rust-toolchain@v1
-      - name: Rust Cache
-        uses: Swatinem/rust-cache@v2
-        with:
-          # Shares across multiple jobs
-          shared-key: "fuzz-test-targets"
       - name: Set Rust Fuzz
         shell: bash
         run: |
           sudo apt-get install -y libfuzzer-14-dev
           rustup install nightly
-          cargo +nightly install cargo-fuzz cargo-gc-bin
+          cargo +nightly install cargo-fuzz cargo-gc-bin --force
       - name: Download pre-built binaries
         uses: actions/download-artifact@v4
         with:
@@ -211,16 +194,11 @@ jobs:
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
       - uses: actions-rust-lang/setup-rust-toolchain@v1
-      - name: Rust Cache
-        uses: Swatinem/rust-cache@v2
-        with:
-          # Shares across multiple jobs
-          shared-key: "fuzz-test-targets"
       - name: Set Rust Fuzz
         shell: bash
         run: |
           sudo apt update && sudo apt install -y libfuzzer-14-dev
-          cargo install cargo-fuzz cargo-gc-bin
+          cargo install cargo-fuzz cargo-gc-bin --force
       - name: Download pre-built binariy
         uses: actions/download-artifact@v4
         with:
@@ -266,13 +244,15 @@ jobs:
         with:
           # Shares across multiple jobs
           shared-key: "build-greptime-ci"
+          cache-all-crates: "true"
+          save-if: ${{ github.ref == 'refs/heads/main' }}
       - name: Install cargo-gc-bin
         shell: bash
-        run: cargo install cargo-gc-bin
+        run: cargo install cargo-gc-bin --force
       - name: Build greptime bianry
         shell: bash
         # `cargo gc` will invoke `cargo build` with specified args
-        run: cargo gc --profile ci -- --bin greptime
+        run: cargo gc --profile ci -- --bin greptime --features pg_kvbackend
       - name: Pack greptime binary
         shell: bash
         run: |
@@ -328,17 +308,12 @@ jobs:
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
       - uses: actions-rust-lang/setup-rust-toolchain@v1
-      - name: Rust Cache
-        uses: Swatinem/rust-cache@v2
-        with:
-          # Shares across multiple jobs
-          shared-key: "fuzz-test-targets"
       - name: Set Rust Fuzz
         shell: bash
         run: |
           sudo apt-get install -y libfuzzer-14-dev
           rustup install nightly
-          cargo +nightly install cargo-fuzz cargo-gc-bin
+          cargo +nightly install cargo-fuzz cargo-gc-bin --force
       # Downloads ci image
       - name: Download pre-built binariy
         uses: actions/download-artifact@v4
@@ -477,17 +452,12 @@ jobs:
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
       - uses: actions-rust-lang/setup-rust-toolchain@v1
-      - name: Rust Cache
-        uses: Swatinem/rust-cache@v2
-        with:
-          # Shares across multiple jobs
-          shared-key: "fuzz-test-targets"
       - name: Set Rust Fuzz
         shell: bash
         run: |
           sudo apt-get install -y libfuzzer-14-dev
           rustup install nightly
-          cargo +nightly install cargo-fuzz cargo-gc-bin
+          cargo +nightly install cargo-fuzz cargo-gc-bin --force
       # Downloads ci image
       - name: Download pre-built binariy
         uses: actions/download-artifact@v4
@@ -589,8 +559,8 @@ jobs:
       - uses: actions/checkout@v4
       - if: matrix.mode.kafka
         name: Setup kafka server
-        working-directory: tests-integration/fixtures/kafka
-        run: docker compose -f docker-compose-standalone.yml up -d --wait
+        working-directory: tests-integration/fixtures
+        run: docker compose up -d --wait kafka
       - name: Download pre-built binaries
         uses: actions/download-artifact@v4
         with:
@@ -620,11 +590,6 @@ jobs:
       - uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
           components: rustfmt
-      - name: Rust Cache
-        uses: Swatinem/rust-cache@v2
-        with:
-          # Shares across multiple jobs
-          shared-key: "check-rust-fmt"
       - name: Check format
         run: make fmt-check
 
@@ -646,11 +611,69 @@ jobs:
           # Shares across multiple jobs
           # Shares with `Check` job
           shared-key: "check-lint"
+          cache-all-crates: "true"
+          save-if: ${{ github.ref == 'refs/heads/main' }}
       - name: Run cargo clippy
         run: make clippy
 
+  conflict-check:
+    name: Check for conflict
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Merge Conflict Finder
+        uses: olivernybroe/action-conflict-finder@v4.0
+
+  test:
+    if: github.event_name != 'merge_group'
+    runs-on: ubuntu-24.04-arm
+    timeout-minutes: 60
+    needs:  [conflict-check, clippy, fmt]
+    steps:
+      - uses: actions/checkout@v4
+      - uses: arduino/setup-protoc@v3
+        with:
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+      - uses: rui314/setup-mold@v1
+      - name: Install toolchain
+        uses: actions-rust-lang/setup-rust-toolchain@v1
+        with:
+            cache: false
+      - name: Rust Cache
+        uses: Swatinem/rust-cache@v2
+        with:
+          # Shares cross multiple jobs
+          shared-key: "coverage-test"
+          cache-all-crates: "true"
+          save-if: ${{ github.ref == 'refs/heads/main' }}
+      - name: Install latest nextest release
+        uses: taiki-e/install-action@nextest
+      - name: Setup external services
+        working-directory: tests-integration/fixtures
+        run: docker compose up -d --wait
+      - name: Run nextest cases
+        run: cargo nextest run --workspace -F dashboard -F pg_kvbackend
+        env:
+          CARGO_BUILD_RUSTFLAGS: "-C link-arg=-fuse-ld=mold"
+          RUST_BACKTRACE: 1
+          CARGO_INCREMENTAL: 0
+          GT_S3_BUCKET: ${{ vars.AWS_CI_TEST_BUCKET }}
+          GT_S3_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }}
+          GT_S3_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }}
+          GT_S3_REGION: ${{ vars.AWS_CI_TEST_BUCKET_REGION }}
+          GT_MINIO_BUCKET: greptime
+          GT_MINIO_ACCESS_KEY_ID: superpower_ci_user
+          GT_MINIO_ACCESS_KEY: superpower_password
+          GT_MINIO_REGION: us-west-2
+          GT_MINIO_ENDPOINT_URL: http://127.0.0.1:9000
+          GT_ETCD_ENDPOINTS: http://127.0.0.1:2379
+          GT_POSTGRES_ENDPOINTS: postgres://greptimedb:admin@127.0.0.1:5432/postgres
+          GT_KAFKA_ENDPOINTS: 127.0.0.1:9092
+          GT_KAFKA_SASL_ENDPOINTS: 127.0.0.1:9093
+          UNITTEST_LOG_DIR: "__unittest_logs"
+
   coverage:
-    if: github.event.pull_request.draft == false
+    if: github.event_name == 'merge_group'
     runs-on: ubuntu-20.04-8-cores
     timeout-minutes: 60
     steps:
@@ -658,48 +681,29 @@ jobs:
       - uses: arduino/setup-protoc@v3
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
-      - uses: KyleMayes/install-llvm-action@v1
-        with:
-          version: "14.0"
+      - uses: rui314/setup-mold@v1
       - name: Install toolchain
         uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
-          components: llvm-tools-preview
+          components: llvm-tools
+          cache: false
       - name: Rust Cache
         uses: Swatinem/rust-cache@v2
         with:
           # Shares cross multiple jobs
           shared-key: "coverage-test"
-      - name: Docker Cache
-        uses: ScribeMD/docker-cache@0.3.7
-        with:
-          key: docker-${{ runner.os }}-coverage
+          save-if: ${{ github.ref == 'refs/heads/main' }}
       - name: Install latest nextest release
         uses: taiki-e/install-action@nextest
       - name: Install cargo-llvm-cov
         uses: taiki-e/install-action@cargo-llvm-cov
-      - name: Install Python
-        uses: actions/setup-python@v5
-        with:
-          python-version: '3.10'
-      - name: Install PyArrow Package
-        run: pip install pyarrow numpy
-      - name: Setup etcd server
-        working-directory: tests-integration/fixtures/etcd
-        run: docker compose -f docker-compose-standalone.yml up -d --wait
-      - name: Setup kafka server
-        working-directory: tests-integration/fixtures/kafka
-        run: docker compose -f docker-compose-standalone.yml up -d --wait
-      - name: Setup minio
-        working-directory: tests-integration/fixtures/minio
-        run: docker compose -f docker-compose-standalone.yml up -d --wait
-      - name: Setup postgres server
-        working-directory: tests-integration/fixtures/postgres
-        run: docker compose -f docker-compose-standalone.yml up -d --wait
+      - name: Setup external services
+        working-directory: tests-integration/fixtures
+        run: docker compose up -d --wait
       - name: Run nextest cases
-        run: cargo llvm-cov nextest --workspace --lcov --output-path lcov.info -F pyo3_backend -F dashboard
+        run: cargo llvm-cov nextest --workspace --lcov --output-path lcov.info -F dashboard -F pg_kvbackend
         env:
-          CARGO_BUILD_RUSTFLAGS: "-C link-arg=-fuse-ld=lld"
+          CARGO_BUILD_RUSTFLAGS: "-C link-arg=-fuse-ld=mold"
           RUST_BACKTRACE: 1
           CARGO_INCREMENTAL: 0
           GT_S3_BUCKET: ${{ vars.AWS_CI_TEST_BUCKET }}

.github/workflows/nightly-build.yml

@@ -27,7 +27,7 @@ on:
       linux_arm64_runner:
         type: choice
         description: The runner uses to build linux-arm64 artifacts
-        default: ec2-c6g.4xlarge-arm64
+        default: ec2-c6g.8xlarge-arm64
         options:
           - ec2-c6g.xlarge-arm64 # 4C8G
           - ec2-c6g.2xlarge-arm64 # 8C16G

.github/workflows/nightly-ci.yml

@@ -1,6 +1,6 @@
 on:
   schedule:
-    - cron: "0 23 * * 1-5"
+    - cron: "0 23 * * 1-4"
   workflow_dispatch:
 
 name: Nightly CI
@@ -91,18 +91,12 @@ jobs:
         uses: Swatinem/rust-cache@v2
       - name: Install Cargo Nextest
         uses: taiki-e/install-action@nextest
-      - name: Install Python
-        uses: actions/setup-python@v5
-        with:
-          python-version: "3.10"
-      - name: Install PyArrow Package
-        run: pip install pyarrow numpy
       - name: Install WSL distribution
         uses: Vampire/setup-wsl@v2
         with:
           distribution: Ubuntu-22.04
       - name: Running tests
-        run: cargo nextest run -F pyo3_backend,dashboard
+        run: cargo nextest run -F dashboard
         env:
           CARGO_BUILD_RUSTFLAGS: "-C linker=lld-link"
           RUST_BACKTRACE: 1
@@ -114,10 +108,55 @@ jobs:
           GT_S3_REGION: ${{ vars.AWS_CI_TEST_BUCKET_REGION }}
           UNITTEST_LOG_DIR: "__unittest_logs"
 
+  ## this is designed for generating cache that usable for pull requests
+  test-on-linux:
+    name: Run tests on Linux
+    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
+    runs-on: ubuntu-20.04-8-cores
+    timeout-minutes: 60
+    steps:
+      - uses: actions/checkout@v4
+      - uses: arduino/setup-protoc@v3
+        with:
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+      - uses: rui314/setup-mold@v1
+      - name: Install Rust toolchain
+        uses: actions-rust-lang/setup-rust-toolchain@v1
+      - name: Rust Cache
+        uses: Swatinem/rust-cache@v2
+        with:
+          # Shares cross multiple jobs
+          shared-key: "coverage-test"
+      - name: Install Cargo Nextest
+        uses: taiki-e/install-action@nextest
+      - name: Setup external services
+        working-directory: tests-integration/fixtures
+        run: docker compose up -d --wait
+      - name: Running tests
+        run: cargo nextest run -F dashboard -F pg_kvbackend
+        env:
+          CARGO_BUILD_RUSTFLAGS: "-C link-arg=-fuse-ld=mold"
+          RUST_BACKTRACE: 1
+          CARGO_INCREMENTAL: 0
+          GT_S3_BUCKET: ${{ vars.AWS_CI_TEST_BUCKET }}
+          GT_S3_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }}
+          GT_S3_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }}
+          GT_S3_REGION: ${{ vars.AWS_CI_TEST_BUCKET_REGION }}
+          GT_MINIO_BUCKET: greptime
+          GT_MINIO_ACCESS_KEY_ID: superpower_ci_user
+          GT_MINIO_ACCESS_KEY: superpower_password
+          GT_MINIO_REGION: us-west-2
+          GT_MINIO_ENDPOINT_URL: http://127.0.0.1:9000
+          GT_ETCD_ENDPOINTS: http://127.0.0.1:2379
+          GT_POSTGRES_ENDPOINTS: postgres://greptimedb:admin@127.0.0.1:5432/postgres
+          GT_KAFKA_ENDPOINTS: 127.0.0.1:9092
+          GT_KAFKA_SASL_ENDPOINTS: 127.0.0.1:9093
+          UNITTEST_LOG_DIR: "__unittest_logs"
+
   cleanbuild-linux-nix:
+    name: Run clean build on Linux
     runs-on: ubuntu-latest-8-cores
     timeout-minutes: 60
-    needs: [coverage, fmt, clippy, check]
     steps:
       - uses: actions/checkout@v4
       - uses: cachix/install-nix-action@v27

.github/workflows/release.yml

@@ -31,7 +31,7 @@ on:
       linux_arm64_runner:
         type: choice
         description: The runner uses to build linux-arm64 artifacts
-        default: ec2-c6g.4xlarge-arm64
+        default: ec2-c6g.8xlarge-arm64
         options:
           - ubuntu-2204-32-cores-arm
           - ec2-c6g.xlarge-arm64 # 4C8G
@@ -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.12.0
+  NEXT_RELEASE_VERSION: v0.11.0
 
 # Permission reference: https://docs.github.com/en/actions/using-jobs/assigning-permissions-to-jobs
 permissions:
@@ -222,18 +222,10 @@ jobs:
             arch: aarch64-apple-darwin
             features: servers/dashboard
             artifacts-dir-prefix: greptime-darwin-arm64
-          - os: ${{ needs.allocate-runners.outputs.macos-runner }}
-            arch: aarch64-apple-darwin
-            features: pyo3_backend,servers/dashboard
-            artifacts-dir-prefix: greptime-darwin-arm64-pyo3
           - os: ${{ needs.allocate-runners.outputs.macos-runner }}
             features: servers/dashboard
             arch: x86_64-apple-darwin
             artifacts-dir-prefix: greptime-darwin-amd64
-          - os: ${{ needs.allocate-runners.outputs.macos-runner }}
-            features: pyo3_backend,servers/dashboard
-            arch: x86_64-apple-darwin
-            artifacts-dir-prefix: greptime-darwin-amd64-pyo3
     runs-on: ${{ matrix.os }}
     outputs:
       build-macos-result: ${{ steps.set-build-macos-result.outputs.build-macos-result }}
@@ -271,10 +263,6 @@ jobs:
             arch: x86_64-pc-windows-msvc
             features: servers/dashboard
             artifacts-dir-prefix: greptime-windows-amd64
-          - os: ${{ needs.allocate-runners.outputs.windows-runner }}
-            arch: x86_64-pc-windows-msvc
-            features: pyo3_backend,servers/dashboard
-            artifacts-dir-prefix: greptime-windows-amd64-pyo3
     runs-on: ${{ matrix.os }}
     outputs:
       build-windows-result: ${{ steps.set-build-windows-result.outputs.build-windows-result }}
@@ -448,6 +436,22 @@ jobs:
           aws-region: ${{ vars.EC2_RUNNER_REGION }}
           github-token: ${{ secrets.GH_PERSONAL_ACCESS_TOKEN }}
 
+  bump-doc-version:
+    name: Bump doc version
+    if: ${{ github.event_name == 'push' || github.event_name == 'schedule' }}
+    needs: [allocate-runners]
+    runs-on: ubuntu-20.04
+    steps:
+      - uses: actions/checkout@v4
+      - uses: ./.github/actions/setup-cyborg
+      - name: Bump doc version
+        working-directory: cyborg
+        run: pnpm tsx bin/bump-doc-version.ts
+        env:
+          VERSION: ${{ needs.allocate-runners.outputs.version }}
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          DOCS_REPO_TOKEN: ${{ secrets.DOCS_REPO_TOKEN }}
+
   notification:
     if: ${{ github.repository == 'GreptimeTeam/greptimedb' && (github.event_name == 'push' || github.event_name == 'schedule') && always() }}
     name: Send notification to Greptime team

Cargo.lock

@@ -188,7 +188,7 @@ checksum = "d301b3b94cb4b2f23d7917810addbbaff90738e0ca2be692bd027e70d7e0330c"
 
 [[package]]
 name = "api"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "common-base",
  "common-decimal",
@@ -730,6 +730,36 @@ version = "1.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"
 
+[[package]]
+name = "attribute-derive"
+version = "0.10.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f1800e974930e9079c965b9ffbcb6667a40401063a26396c7b4f15edc92da690"
+dependencies = [
+ "attribute-derive-macro",
+ "derive-where",
+ "manyhow",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.90",
+]
+
+[[package]]
+name = "attribute-derive-macro"
+version = "0.10.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5d908eb786ef94296bff86f90130b3b748b49401dc81fd2bb8b3dccd44cfacbd"
+dependencies = [
+ "collection_literals",
+ "interpolator",
+ "manyhow",
+ "proc-macro-utils",
+ "proc-macro2",
+ "quote",
+ "quote-use",
+ "syn 2.0.90",
+]
+
 [[package]]
 name = "atty"
 version = "0.2.14"
@@ -743,7 +773,7 @@ dependencies = [
 
 [[package]]
 name = "auth"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "async-trait",
@@ -1284,7 +1314,7 @@ dependencies = [
 
 [[package]]
 name = "cache"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "catalog",
  "common-error",
@@ -1318,7 +1348,7 @@ checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5"
 
 [[package]]
 name = "catalog"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "arrow",
@@ -1654,7 +1684,7 @@ checksum = "1462739cb27611015575c0c11df5df7601141071f07518d56fcc1be504cbec97"
 
 [[package]]
 name = "cli"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "async-trait",
  "auth",
@@ -1697,7 +1727,7 @@ dependencies = [
  "session",
  "snafu 0.8.5",
  "store-api",
- "substrait 0.12.0",
+ "substrait 0.11.3",
  "table",
  "tempfile",
  "tokio",
@@ -1706,7 +1736,7 @@ dependencies = [
 
 [[package]]
 name = "client"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "arc-swap",
@@ -1733,7 +1763,7 @@ dependencies = [
  "rand",
  "serde_json",
  "snafu 0.8.5",
- "substrait 0.12.0",
+ "substrait 0.11.3",
  "substrait 0.37.3",
  "tokio",
  "tokio-stream",
@@ -1774,7 +1804,7 @@ dependencies = [
 
 [[package]]
 name = "cmd"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "async-trait",
  "auth",
@@ -1834,7 +1864,7 @@ dependencies = [
  "similar-asserts",
  "snafu 0.8.5",
  "store-api",
- "substrait 0.12.0",
+ "substrait 0.11.3",
  "table",
  "temp-env",
  "tempfile",
@@ -1845,6 +1875,12 @@ dependencies = [
  "tracing-appender",
 ]
 
+[[package]]
+name = "collection_literals"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "186dce98367766de751c42c4f03970fc60fc012296e706ccbb9d5df9b6c1e271"
+
 [[package]]
 name = "colorchoice"
 version = "1.0.2"
@@ -1880,7 +1916,7 @@ checksum = "55b672471b4e9f9e95499ea597ff64941a309b2cdbffcc46f2cc5e2d971fd335"
 
 [[package]]
 name = "common-base"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "anymap2",
  "async-trait",
@@ -1902,11 +1938,11 @@ dependencies = [
 
 [[package]]
 name = "common-catalog"
-version = "0.12.0"
+version = "0.11.3"
 
 [[package]]
 name = "common-config"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "common-base",
  "common-error",
@@ -1929,7 +1965,7 @@ dependencies = [
 
 [[package]]
 name = "common-datasource"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "arrow",
  "arrow-schema",
@@ -1965,7 +2001,7 @@ dependencies = [
 
 [[package]]
 name = "common-decimal"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "bigdecimal 0.4.5",
  "common-error",
@@ -1978,8 +2014,9 @@ dependencies = [
 
 [[package]]
 name = "common-error"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
+ "http 0.2.12",
  "snafu 0.8.5",
  "strum 0.25.0",
  "tonic 0.11.0",
@@ -1987,7 +2024,7 @@ dependencies = [
 
 [[package]]
 name = "common-frontend"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "async-trait",
  "common-error",
@@ -1997,7 +2034,7 @@ dependencies = [
 
 [[package]]
 name = "common-function"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "approx 0.5.1",
@@ -2041,7 +2078,7 @@ dependencies = [
 
 [[package]]
 name = "common-greptimedb-telemetry"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "async-trait",
  "common-runtime",
@@ -2058,7 +2095,7 @@ dependencies = [
 
 [[package]]
 name = "common-grpc"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "arrow-flight",
@@ -2084,7 +2121,7 @@ dependencies = [
 
 [[package]]
 name = "common-grpc-expr"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "common-base",
@@ -2103,7 +2140,7 @@ dependencies = [
 
 [[package]]
 name = "common-macro"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "arc-swap",
  "common-query",
@@ -2117,7 +2154,7 @@ dependencies = [
 
 [[package]]
 name = "common-mem-prof"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "common-error",
  "common-macro",
@@ -2130,7 +2167,7 @@ dependencies = [
 
 [[package]]
 name = "common-meta"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "anymap2",
  "api",
@@ -2187,7 +2224,7 @@ dependencies = [
 
 [[package]]
 name = "common-options"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "common-grpc",
  "humantime-serde",
@@ -2196,11 +2233,11 @@ dependencies = [
 
 [[package]]
 name = "common-plugins"
-version = "0.12.0"
+version = "0.11.3"
 
 [[package]]
 name = "common-pprof"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "common-error",
  "common-macro",
@@ -2212,7 +2249,7 @@ dependencies = [
 
 [[package]]
 name = "common-procedure"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "async-stream",
  "async-trait",
@@ -2239,7 +2276,7 @@ dependencies = [
 
 [[package]]
 name = "common-procedure-test"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "async-trait",
  "common-procedure",
@@ -2247,7 +2284,7 @@ dependencies = [
 
 [[package]]
 name = "common-query"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "async-trait",
@@ -2273,7 +2310,7 @@ dependencies = [
 
 [[package]]
 name = "common-recordbatch"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "arc-swap",
  "common-error",
@@ -2292,7 +2329,7 @@ dependencies = [
 
 [[package]]
 name = "common-runtime"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "async-trait",
  "clap 4.5.19",
@@ -2314,13 +2351,15 @@ dependencies = [
  "snafu 0.8.5",
  "tempfile",
  "tokio",
+ "tokio-metrics",
+ "tokio-metrics-collector",
  "tokio-test",
  "tokio-util",
 ]
 
 [[package]]
 name = "common-telemetry"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "atty",
  "backtrace",
@@ -2348,7 +2387,7 @@ dependencies = [
 
 [[package]]
 name = "common-test-util"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "client",
  "common-query",
@@ -2360,7 +2399,7 @@ dependencies = [
 
 [[package]]
 name = "common-time"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "arrow",
  "chrono",
@@ -2378,7 +2417,7 @@ dependencies = [
 
 [[package]]
 name = "common-version"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "build-data",
  "const_format",
@@ -2388,7 +2427,7 @@ dependencies = [
 
 [[package]]
 name = "common-wal"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "common-base",
  "common-error",
@@ -3187,7 +3226,7 @@ dependencies = [
 
 [[package]]
 name = "datanode"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "arrow-flight",
@@ -3238,7 +3277,7 @@ dependencies = [
  "session",
  "snafu 0.8.5",
  "store-api",
- "substrait 0.12.0",
+ "substrait 0.11.3",
  "table",
  "tokio",
  "toml 0.8.19",
@@ -3247,7 +3286,7 @@ dependencies = [
 
 [[package]]
 name = "datatypes"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -3346,6 +3385,17 @@ dependencies = [
  "syn 2.0.90",
 ]
 
+[[package]]
+name = "derive-where"
+version = "1.2.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "62d671cc41a825ebabc75757b62d3d168c577f9149b2d49ece1dad1f72119d25"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.90",
+]
+
 [[package]]
 name = "derive_arbitrary"
 version = "1.3.2"
@@ -3860,7 +3910,7 @@ dependencies = [
 
 [[package]]
 name = "file-engine"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "async-trait",
@@ -3976,7 +4026,7 @@ checksum = "8bf7cc16383c4b8d58b9905a8509f02926ce3058053c056376248d958c9df1e8"
 
 [[package]]
 name = "flow"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "arrow",
@@ -4011,7 +4061,10 @@ dependencies = [
  "enum-as-inner",
  "enum_dispatch",
  "futures",
+ "get-size-derive2",
+ "get-size2",
  "greptime-proto",
+ "http 0.2.12",
  "hydroflow",
  "itertools 0.10.5",
  "lazy_static",
@@ -4032,7 +4085,7 @@ dependencies = [
  "snafu 0.8.5",
  "store-api",
  "strum 0.25.0",
- "substrait 0.12.0",
+ "substrait 0.11.3",
  "table",
  "tokio",
  "tonic 0.11.0",
@@ -4070,7 +4123,7 @@ checksum = "6c2141d6d6c8512188a7891b4b01590a45f6dac67afb4f255c4124dbb86d4eaa"
 
 [[package]]
 name = "frontend"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "arc-swap",
@@ -4103,6 +4156,7 @@ dependencies = [
  "futures",
  "humantime-serde",
  "lazy_static",
+ "log-query",
  "log-store",
  "meta-client",
  "opentelemetry-proto 0.5.0",
@@ -4415,6 +4469,23 @@ dependencies = [
  "libm",
 ]
 
+[[package]]
+name = "get-size-derive2"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1fd26d3a97ea14d289c8b54180243ecfe465f3fa9c279a6336d7a003698fc39d"
+dependencies = [
+ "attribute-derive",
+ "quote",
+ "syn 2.0.90",
+]
+
+[[package]]
+name = "get-size2"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "159c430715e540d2198fa981d39cd45563ccc60900de187f5b152b33b1cb408e"
+
 [[package]]
 name = "gethostname"
 version = "0.2.3"
@@ -4487,7 +4558,7 @@ dependencies = [
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=a875e976441188028353f7274a46a7e6e065c5d4#a875e976441188028353f7274a46a7e6e065c5d4"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=43ddd8dea69f4df0fe2e8b5cdc0044d2cfa35908#43ddd8dea69f4df0fe2e8b5cdc0044d2cfa35908"
 dependencies = [
  "prost 0.12.6",
  "serde",
@@ -5202,7 +5273,7 @@ dependencies = [
 
 [[package]]
 name = "index"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "async-trait",
  "asynchronous-codec",
@@ -5219,6 +5290,7 @@ dependencies = [
  "futures",
  "greptime-proto",
  "mockall",
+ "parquet",
  "pin-project",
  "prost 0.12.6",
  "rand",
@@ -5346,6 +5418,12 @@ version = "4.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0d762194228a2f1c11063e46e32e5acb96e66e906382b9eb5441f2e0504bbd5a"
 
+[[package]]
+name = "interpolator"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "71dd52191aae121e8611f1e8dc3e324dd0dd1dee1e6dd91d10ee07a3cfb4d9d8"
+
 [[package]]
 name = "inventory"
 version = "0.3.15"
@@ -6045,18 +6123,19 @@ checksum = "a7a70ba024b9dc04c27ea2f0c0548feb474ec5c54bba33a7f72f873a39d07b24"
 
 [[package]]
 name = "log-query"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "chrono",
  "common-error",
  "common-macro",
+ "serde",
  "snafu 0.8.5",
  "table",
 ]
 
 [[package]]
 name = "log-store"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "async-stream",
  "async-trait",
@@ -6244,6 +6323,29 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "manyhow"
+version = "0.11.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b33efb3ca6d3b07393750d4030418d594ab1139cee518f0dc88db70fec873587"
+dependencies = [
+ "manyhow-macros",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.90",
+]
+
+[[package]]
+name = "manyhow-macros"
+version = "0.11.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "46fce34d199b78b6e6073abf984c9cf5fd3e9330145a93ee0738a7443e371495"
+dependencies = [
+ "proc-macro-utils",
+ "proc-macro2",
+ "quote",
+]
+
 [[package]]
 name = "maplit"
 version = "1.0.2"
@@ -6377,7 +6479,7 @@ dependencies = [
 
 [[package]]
 name = "meta-client"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "async-trait",
@@ -6404,7 +6506,7 @@ dependencies = [
 
 [[package]]
 name = "meta-srv"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "async-trait",
@@ -6483,7 +6585,7 @@ dependencies = [
 
 [[package]]
 name = "metric-engine"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "aquamarine",
@@ -6577,7 +6679,7 @@ dependencies = [
 
 [[package]]
 name = "mito2"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "aquamarine",
@@ -7314,7 +7416,7 @@ dependencies = [
 
 [[package]]
 name = "object-store"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "anyhow",
  "bytes",
@@ -7375,8 +7477,7 @@ checksum = "b410bbe7e14ab526a0e86877eb47c6996a2bd7746f027ba551028c925390e4e9"
 [[package]]
 name = "opendal"
 version = "0.50.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cb28bb6c64e116ceaf8dd4e87099d3cfea4a58e85e62b104fef74c91afba0f44"
+source = "git+https://github.com/GreptimeTeam/opendal.git?rev=c82605177f2feec83e49dcaa537c505639d94024#c82605177f2feec83e49dcaa537c505639d94024"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -7568,7 +7669,7 @@ dependencies = [
 
 [[package]]
 name = "operator"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "ahash 0.8.11",
  "api",
@@ -7616,7 +7717,7 @@ dependencies = [
  "sql",
  "sqlparser 0.45.0 (git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=54a267ac89c09b11c0c88934690530807185d3e7)",
  "store-api",
- "substrait 0.12.0",
+ "substrait 0.11.3",
  "table",
  "tokio",
  "tokio-util",
@@ -7866,7 +7967,7 @@ dependencies = [
 
 [[package]]
 name = "partition"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "async-trait",
@@ -8065,7 +8166,7 @@ dependencies = [
  "rand",
  "ring 0.17.8",
  "rust_decimal",
- "thiserror 2.0.4",
+ "thiserror 2.0.6",
  "tokio",
  "tokio-rustls 0.26.0",
  "tokio-util",
@@ -8152,7 +8253,7 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
 
 [[package]]
 name = "pipeline"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "ahash 0.8.11",
  "api",
@@ -8314,7 +8415,7 @@ dependencies = [
 
 [[package]]
 name = "plugins"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "auth",
  "clap 4.5.19",
@@ -8528,6 +8629,17 @@ dependencies = [
  "version_check",
 ]
 
+[[package]]
+name = "proc-macro-utils"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eeaf08a13de400bc215877b5bdc088f241b12eb42f0a548d3390dc1c56bb7071"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "smallvec",
+]
+
 [[package]]
 name = "proc-macro2"
 version = "1.0.92"
@@ -8591,7 +8703,7 @@ dependencies = [
 
 [[package]]
 name = "promql"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "ahash 0.8.11",
  "async-trait",
@@ -8826,7 +8938,7 @@ dependencies = [
 
 [[package]]
 name = "puffin"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "async-compression 0.4.13",
  "async-trait",
@@ -8951,7 +9063,7 @@ dependencies = [
 
 [[package]]
 name = "query"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "ahash 0.8.11",
  "api",
@@ -8992,8 +9104,10 @@ dependencies = [
  "humantime",
  "itertools 0.10.5",
  "lazy_static",
+ "log-query",
  "meter-core",
  "meter-macros",
+ "nalgebra 0.33.2",
  "num",
  "num-traits",
  "object-store",
@@ -9014,7 +9128,7 @@ dependencies = [
  "sqlparser 0.45.0 (git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=54a267ac89c09b11c0c88934690530807185d3e7)",
  "statrs",
  "store-api",
- "substrait 0.12.0",
+ "substrait 0.11.3",
  "table",
  "tokio",
  "tokio-stream",
@@ -9107,6 +9221,28 @@ dependencies = [
  "proc-macro2",
 ]
 
+[[package]]
+name = "quote-use"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9619db1197b497a36178cfc736dc96b271fe918875fbf1344c436a7e93d0321e"
+dependencies = [
+ "quote",
+ "quote-use-macros",
+]
+
+[[package]]
+name = "quote-use-macros"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "82ebfb7faafadc06a7ab141a6f67bcfb24cb8beb158c6fe933f2f035afa99f35"
+dependencies = [
+ "proc-macro-utils",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.90",
+]
+
 [[package]]
 name = "radium"
 version = "0.7.0"
@@ -10476,7 +10612,7 @@ checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
 
 [[package]]
 name = "script"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "arc-swap",
@@ -10768,7 +10904,7 @@ dependencies = [
 
 [[package]]
 name = "servers"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "ahash 0.8.11",
  "api",
@@ -10824,6 +10960,7 @@ dependencies = [
  "json5",
  "jsonb",
  "lazy_static",
+ "log-query",
  "loki-api",
  "mime_guess",
  "mysql_async",
@@ -10879,7 +11016,7 @@ dependencies = [
 
 [[package]]
 name = "session"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "arc-swap",
@@ -11233,7 +11370,7 @@ dependencies = [
 
 [[package]]
 name = "sql"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "chrono",
@@ -11297,7 +11434,7 @@ dependencies = [
 
 [[package]]
 name = "sqlness-runner"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "async-trait",
  "clap 4.5.19",
@@ -11515,7 +11652,7 @@ dependencies = [
 
 [[package]]
 name = "store-api"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "aquamarine",
@@ -11677,7 +11814,7 @@ dependencies = [
 
 [[package]]
 name = "substrait"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "async-trait",
  "bytes",
@@ -11876,7 +12013,7 @@ dependencies = [
 
 [[package]]
 name = "table"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "async-trait",
@@ -12153,7 +12290,7 @@ checksum = "3369f5ac52d5eb6ab48c6b4ffdc8efbcad6b89c765749064ba298f2c68a16a76"
 
 [[package]]
 name = "tests-fuzz"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "arbitrary",
  "async-trait",
@@ -12196,7 +12333,7 @@ dependencies = [
 
 [[package]]
 name = "tests-integration"
-version = "0.12.0"
+version = "0.11.3"
 dependencies = [
  "api",
  "arrow-flight",
@@ -12236,6 +12373,7 @@ dependencies = [
  "futures-util",
  "hex",
  "itertools 0.10.5",
+ "log-query",
  "loki-api",
  "meta-client",
  "meta-srv",
@@ -12260,7 +12398,7 @@ dependencies = [
  "sql",
  "sqlx",
  "store-api",
- "substrait 0.12.0",
+ "substrait 0.11.3",
  "table",
  "tempfile",
  "time",
@@ -12306,11 +12444,11 @@ dependencies = [
 
 [[package]]
 name = "thiserror"
-version = "2.0.4"
+version = "2.0.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2f49a1853cf82743e3b7950f77e0f4d622ca36cf4317cba00c767838bac8d490"
+checksum = "8fec2a1820ebd077e2b90c4df007bebf344cd394098a13c563957d0afc83ea47"
 dependencies = [
- "thiserror-impl 2.0.4",
+ "thiserror-impl 2.0.6",
 ]
 
 [[package]]
@@ -12326,9 +12464,9 @@ dependencies = [
 
 [[package]]
 name = "thiserror-impl"
-version = "2.0.4"
+version = "2.0.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8381894bb3efe0c4acac3ded651301ceee58a15d47c2e34885ed1908ad667061"
+checksum = "d65750cab40f4ff1929fb1ba509e9914eb756131cef4210da8d5d700d26f6312"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -12484,9 +12622,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
 
 [[package]]
 name = "tokio"
-version = "1.40.0"
+version = "1.42.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e2b070231665d27ad9ec9b8df639893f46727666c6767db40317fbe920a5d998"
+checksum = "5cec9b21b0450273377fc97bd4c33a8acffc8c996c987a7c5b319a0083707551"
 dependencies = [
  "backtrace",
  "bytes",
@@ -12522,6 +12660,31 @@ dependencies = [
  "syn 2.0.90",
 ]
 
+[[package]]
+name = "tokio-metrics"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eace09241d62c98b7eeb1107d4c5c64ca3bd7da92e8c218c153ab3a78f9be112"
+dependencies = [
+ "futures-util",
+ "pin-project-lite",
+ "tokio",
+ "tokio-stream",
+]
+
+[[package]]
+name = "tokio-metrics-collector"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a8092b7a97ed5dac2f44892db190eca8f476ede0fa585bc87664de4151cd0b64"
+dependencies = [
+ "lazy_static",
+ "parking_lot 0.12.3",
+ "prometheus",
+ "tokio",
+ "tokio-metrics",
+]
+
 [[package]]
 name = "tokio-postgres"
 version = "0.7.12"

Cargo.toml

@@ -68,7 +68,7 @@ members = [
 resolver = "2"
 
 [workspace.package]
-version = "0.12.0"
+version = "0.11.3"
 edition = "2021"
 license = "Apache-2.0"
 
@@ -124,8 +124,9 @@ etcd-client = "0.13"
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "a875e976441188028353f7274a46a7e6e065c5d4" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "43ddd8dea69f4df0fe2e8b5cdc0044d2cfa35908" }
 hex = "0.4"
+http = "0.2"
 humantime = "2.1"
 humantime-serde = "1.1"
 itertools = "0.10"
@@ -134,6 +135,7 @@ lazy_static = "1.4"
 meter-core = { git = "https://github.com/GreptimeTeam/greptime-meter.git", rev = "a10facb353b41460eeb98578868ebf19c2084fac" }
 mockall = "0.11.4"
 moka = "0.12"
+nalgebra = "0.33"
 notify = "6.1"
 num_cpus = "1.16"
 once_cell = "1.18"
@@ -238,6 +240,7 @@ file-engine = { path = "src/file-engine" }
 flow = { path = "src/flow" }
 frontend = { path = "src/frontend", default-features = false }
 index = { path = "src/index" }
+log-query = { path = "src/log-query" }
 log-store = { path = "src/log-store" }
 meta-client = { path = "src/meta-client" }
 meta-srv = { path = "src/meta-srv" }

README.md

@@ -70,23 +70,23 @@ Our core developers have been building time-series data platforms for years. Bas
 
 * **Unified Processing of Metrics, Logs, and Events**
 
-GreptimeDB unifies time series data processing by treating all data - whether metrics, logs, or events - as timestamped events with context. Users can analyze this data using either [SQL](https://docs.greptime.com/user-guide/query-data/sql) or [PromQL](https://docs.greptime.com/user-guide/query-data/promql) and leverage stream processing ([Flow](https://docs.greptime.com/user-guide/continuous-aggregation/overview)) to enable continuous aggregation. [Read more](https://docs.greptime.com/user-guide/concepts/data-model).
+  GreptimeDB unifies time series data processing by treating all data - whether metrics, logs, or events - as timestamped events with context. Users can analyze this data using either [SQL](https://docs.greptime.com/user-guide/query-data/sql) or [PromQL](https://docs.greptime.com/user-guide/query-data/promql) and leverage stream processing ([Flow](https://docs.greptime.com/user-guide/flow-computation/overview)) to enable continuous aggregation. [Read more](https://docs.greptime.com/user-guide/concepts/data-model).
 
 * **Cloud-native Distributed Database**
 
-Built for [Kubernetes](https://docs.greptime.com/user-guide/deployments/deploy-on-kubernetes/greptimedb-operator-management). GreptimeDB achieves seamless scalability with its [cloud-native architecture](https://docs.greptime.com/user-guide/concepts/architecture) of separated compute and storage, built on object storage (AWS S3, Azure Blob Storage, etc.) while enabling cross-cloud deployment through a unified data access layer.
+  Built for [Kubernetes](https://docs.greptime.com/user-guide/deployments/deploy-on-kubernetes/greptimedb-operator-management). GreptimeDB achieves seamless scalability with its [cloud-native architecture](https://docs.greptime.com/user-guide/concepts/architecture) of separated compute and storage, built on object storage (AWS S3, Azure Blob Storage, etc.) while enabling cross-cloud deployment through a unified data access layer.
 
 * **Performance and Cost-effective**
 
-Written in pure Rust for superior performance and reliability. GreptimeDB features a distributed query engine with intelligent indexing to handle high cardinality data efficiently. Its optimized columnar storage achieves 50x cost efficiency on cloud object storage through advanced compression. [Benchmark reports](https://www.greptime.com/blogs/2024-09-09-report-summary).
+  Written in pure Rust for superior performance and reliability. GreptimeDB features a distributed query engine with intelligent indexing to handle high cardinality data efficiently. Its optimized columnar storage achieves 50x cost efficiency on cloud object storage through advanced compression. [Benchmark reports](https://www.greptime.com/blogs/2024-09-09-report-summary).
 
 * **Cloud-Edge Collaboration**
 
-GreptimeDB seamlessly operates across cloud and edge (ARM/Android/Linux), providing consistent APIs and control plane for unified data management and efficient synchronization. [Learn how to run on Android](https://docs.greptime.com/user-guide/deployments/run-on-android/).
+  GreptimeDB seamlessly operates across cloud and edge (ARM/Android/Linux), providing consistent APIs and control plane for unified data management and efficient synchronization. [Learn how to run on Android](https://docs.greptime.com/user-guide/deployments/run-on-android/).
 
 * **Multi-protocol Ingestion, SQL & PromQL Ready**
 
-Widely adopted database protocols and APIs, including MySQL, PostgreSQL, InfluxDB, OpenTelemetry, Loki and Prometheus, etc.  Effortless Adoption & Seamless Migration. [Supported Protocols Overview](https://docs.greptime.com/user-guide/protocols/overview).
+  Widely adopted database protocols and APIs, including MySQL, PostgreSQL, InfluxDB, OpenTelemetry, Loki and Prometheus, etc.  Effortless Adoption & Seamless Migration. [Supported Protocols Overview](https://docs.greptime.com/user-guide/protocols/overview).
 
 For more detailed info please read  [Why GreptimeDB](https://docs.greptime.com/user-guide/concepts/why-greptimedb).
 
@@ -138,7 +138,7 @@ Check the prerequisite:
 
 * [Rust toolchain](https://www.rust-lang.org/tools/install) (nightly)
 * [Protobuf compiler](https://grpc.io/docs/protoc-installation/) (>= 3.15)
-* Python toolchain (optional): Required only if built with PyO3 backend. More detail for compiling with PyO3 can be found in its [documentation](https://pyo3.rs/v0.18.1/building_and_distribution#configuring-the-python-version).
+* Python toolchain (optional): Required only if built with PyO3 backend. More details for compiling with PyO3 can be found in its [documentation](https://pyo3.rs/v0.18.1/building_and_distribution#configuring-the-python-version).
 
 Build GreptimeDB binary:
 
@@ -154,6 +154,10 @@ cargo run -- standalone start
 
 ## Tools & Extensions
 
+### Kubernetes
+
+- [GreptimeDB Operator](https://github.com/GrepTimeTeam/greptimedb-operator)
+
 ### Dashboard
 
 - [The dashboard UI for GreptimeDB](https://github.com/GreptimeTeam/dashboard)
@@ -173,7 +177,7 @@ Our official Grafana dashboard for monitoring GreptimeDB is available at [grafan
 
 ## Project Status
 
-GreptimeDB is currently in Beta. We are targeting GA (General Availability) with v1.0 release by Early 2025. 
+GreptimeDB is currently in Beta. We are targeting GA (General Availability) with v1.0 release by Early 2025.
 
 While in Beta, GreptimeDB is already:
 

config/config.md

@@ -18,6 +18,7 @@
 | `init_regions_parallelism` | Integer | `16` | Parallelism of initializing regions. |
 | `max_concurrent_queries` | Integer | `0` | The maximum current queries allowed to be executed. Zero means unlimited. |
 | `enable_telemetry` | Bool | `true` | Enable telemetry to collect anonymous usage data. Enabled by default. |
+| `max_in_flight_write_bytes` | String | Unset | The maximum in-flight write bytes. |
 | `runtime` | -- | -- | The runtime options. |
 | `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
 | `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute the runtime for global write operations. |
@@ -150,12 +151,17 @@
 | `region_engine.mito.inverted_index.intermediate_path` | String | `""` | Deprecated, use `region_engine.mito.index.aux_path` instead. |
 | `region_engine.mito.inverted_index.metadata_cache_size` | String | `64MiB` | Cache size for inverted index metadata. |
 | `region_engine.mito.inverted_index.content_cache_size` | String | `128MiB` | Cache size for inverted index content. |
-| `region_engine.mito.inverted_index.content_cache_page_size` | String | `8MiB` | Page size for inverted index content cache. |
+| `region_engine.mito.inverted_index.content_cache_page_size` | String | `64KiB` | Page size for inverted index content cache. |
 | `region_engine.mito.fulltext_index` | -- | -- | The options for full-text index in Mito engine. |
 | `region_engine.mito.fulltext_index.create_on_flush` | String | `auto` | Whether to create the index on flush.<br/>- `auto`: automatically (default)<br/>- `disable`: never |
 | `region_engine.mito.fulltext_index.create_on_compaction` | String | `auto` | Whether to create the index on compaction.<br/>- `auto`: automatically (default)<br/>- `disable`: never |
 | `region_engine.mito.fulltext_index.apply_on_query` | String | `auto` | Whether to apply the index on query<br/>- `auto`: automatically (default)<br/>- `disable`: never |
 | `region_engine.mito.fulltext_index.mem_threshold_on_create` | String | `auto` | Memory threshold for index creation.<br/>- `auto`: automatically determine the threshold based on the system memory size (default)<br/>- `unlimited`: no memory limit<br/>- `[size]` e.g. `64MB`: fixed memory threshold |
+| `region_engine.mito.bloom_filter_index` | -- | -- | The options for bloom filter in Mito engine. |
+| `region_engine.mito.bloom_filter_index.create_on_flush` | String | `auto` | Whether to create the bloom filter on flush.<br/>- `auto`: automatically (default)<br/>- `disable`: never |
+| `region_engine.mito.bloom_filter_index.create_on_compaction` | String | `auto` | Whether to create the bloom filter on compaction.<br/>- `auto`: automatically (default)<br/>- `disable`: never |
+| `region_engine.mito.bloom_filter_index.apply_on_query` | String | `auto` | Whether to apply the bloom filter on query<br/>- `auto`: automatically (default)<br/>- `disable`: never |
+| `region_engine.mito.bloom_filter_index.mem_threshold_on_create` | String | `auto` | Memory threshold for bloom filter creation.<br/>- `auto`: automatically determine the threshold based on the system memory size (default)<br/>- `unlimited`: no memory limit<br/>- `[size]` e.g. `64MB`: fixed memory threshold |
 | `region_engine.mito.memtable` | -- | -- | -- |
 | `region_engine.mito.memtable.type` | String | `time_series` | Memtable type.<br/>- `time_series`: time-series memtable<br/>- `partition_tree`: partition tree memtable (experimental) |
 | `region_engine.mito.memtable.index_max_keys_per_shard` | Integer | `8192` | The max number of keys in one shard.<br/>Only available for `partition_tree` memtable. |
@@ -195,6 +201,7 @@
 | Key | Type | Default | Descriptions |
 | --- | -----| ------- | ----------- |
 | `default_timezone` | String | Unset | The default timezone of the server. |
+| `max_in_flight_write_bytes` | String | Unset | The maximum in-flight write bytes. |
 | `runtime` | -- | -- | The runtime options. |
 | `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
 | `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute the runtime for global write operations. |
@@ -421,7 +428,7 @@
 | `storage` | -- | -- | The data storage options. |
 | `storage.data_home` | String | `/tmp/greptimedb/` | The working home directory. |
 | `storage.type` | String | `File` | The storage type used to store the data.<br/>- `File`: the data is stored in the local file system.<br/>- `S3`: the data is stored in the S3 object storage.<br/>- `Gcs`: the data is stored in the Google Cloud Storage.<br/>- `Azblob`: the data is stored in the Azure Blob Storage.<br/>- `Oss`: the data is stored in the Aliyun OSS. |
-| `storage.cache_path` | String | Unset | Read cache configuration for object storage such as 'S3' etc, it's configured by default when using object storage. It is recommended to configure it when using object storage for better performance.<br/>A local file directory, defaults to `{data_home}/object_cache/read`. An empty string means disabling. |
+| `storage.cache_path` | String | Unset | Read cache configuration for object storage such as 'S3' etc, it's configured by default when using object storage. It is recommended to configure it when using object storage for better performance.<br/>A local file directory, defaults to `{data_home}`. An empty string means disabling. |
 | `storage.cache_capacity` | String | Unset | The local file cache capacity in bytes. If your disk space is sufficient, it is recommended to set it larger. |
 | `storage.bucket` | String | Unset | The S3 bucket name.<br/>**It's only used when the storage type is `S3`, `Oss` and `Gcs`**. |
 | `storage.root` | String | Unset | The S3 data will be stored in the specified prefix, for example, `s3://${bucket}/${root}`.<br/>**It's only used when the storage type is `S3`, `Oss` and `Azblob`**. |
@@ -460,7 +467,7 @@
 | `region_engine.mito.page_cache_size` | String | Auto | Cache size for pages of SST row groups. Setting it to 0 to disable the cache.<br/>If not set, it's default to 1/8 of OS memory. |
 | `region_engine.mito.selector_result_cache_size` | String | Auto | Cache size for time series selector (e.g. `last_value()`). Setting it to 0 to disable the cache.<br/>If not set, it's default to 1/16 of OS memory with a max limitation of 512MB. |
 | `region_engine.mito.enable_experimental_write_cache` | Bool | `false` | Whether to enable the experimental write cache, it's enabled by default when using object storage. It is recommended to enable it when using object storage for better performance. |
-| `region_engine.mito.experimental_write_cache_path` | String | `""` | File system path for write cache, defaults to `{data_home}/object_cache/write`. |
+| `region_engine.mito.experimental_write_cache_path` | String | `""` | File system path for write cache, defaults to `{data_home}`. |
 | `region_engine.mito.experimental_write_cache_size` | String | `5GiB` | Capacity for write cache. If your disk space is sufficient, it is recommended to set it larger. |
 | `region_engine.mito.experimental_write_cache_ttl` | String | Unset | TTL for write cache. |
 | `region_engine.mito.sst_write_buffer_size` | String | `8MB` | Buffer size for SST writing. |
@@ -478,12 +485,17 @@
 | `region_engine.mito.inverted_index.intermediate_path` | String | `""` | Deprecated, use `region_engine.mito.index.aux_path` instead. |
 | `region_engine.mito.inverted_index.metadata_cache_size` | String | `64MiB` | Cache size for inverted index metadata. |
 | `region_engine.mito.inverted_index.content_cache_size` | String | `128MiB` | Cache size for inverted index content. |
-| `region_engine.mito.inverted_index.content_cache_page_size` | String | `8MiB` | Page size for inverted index content cache. |
+| `region_engine.mito.inverted_index.content_cache_page_size` | String | `64KiB` | Page size for inverted index content cache. |
 | `region_engine.mito.fulltext_index` | -- | -- | The options for full-text index in Mito engine. |
 | `region_engine.mito.fulltext_index.create_on_flush` | String | `auto` | Whether to create the index on flush.<br/>- `auto`: automatically (default)<br/>- `disable`: never |
 | `region_engine.mito.fulltext_index.create_on_compaction` | String | `auto` | Whether to create the index on compaction.<br/>- `auto`: automatically (default)<br/>- `disable`: never |
 | `region_engine.mito.fulltext_index.apply_on_query` | String | `auto` | Whether to apply the index on query<br/>- `auto`: automatically (default)<br/>- `disable`: never |
 | `region_engine.mito.fulltext_index.mem_threshold_on_create` | String | `auto` | Memory threshold for index creation.<br/>- `auto`: automatically determine the threshold based on the system memory size (default)<br/>- `unlimited`: no memory limit<br/>- `[size]` e.g. `64MB`: fixed memory threshold |
+| `region_engine.mito.bloom_filter_index` | -- | -- | The options for bloom filter index in Mito engine. |
+| `region_engine.mito.bloom_filter_index.create_on_flush` | String | `auto` | Whether to create the index on flush.<br/>- `auto`: automatically (default)<br/>- `disable`: never |
+| `region_engine.mito.bloom_filter_index.create_on_compaction` | String | `auto` | Whether to create the index on compaction.<br/>- `auto`: automatically (default)<br/>- `disable`: never |
+| `region_engine.mito.bloom_filter_index.apply_on_query` | String | `auto` | Whether to apply the index on query<br/>- `auto`: automatically (default)<br/>- `disable`: never |
+| `region_engine.mito.bloom_filter_index.mem_threshold_on_create` | String | `auto` | Memory threshold for the index creation.<br/>- `auto`: automatically determine the threshold based on the system memory size (default)<br/>- `unlimited`: no memory limit<br/>- `[size]` e.g. `64MB`: fixed memory threshold |
 | `region_engine.mito.memtable` | -- | -- | -- |
 | `region_engine.mito.memtable.type` | String | `time_series` | Memtable type.<br/>- `time_series`: time-series memtable<br/>- `partition_tree`: partition tree memtable (experimental) |
 | `region_engine.mito.memtable.index_max_keys_per_shard` | Integer | `8192` | The max number of keys in one shard.<br/>Only available for `partition_tree` memtable. |

config/datanode.example.toml

@@ -294,7 +294,7 @@ data_home = "/tmp/greptimedb/"
 type = "File"
 
 ## Read cache configuration for object storage such as 'S3' etc, it's configured by default when using object storage. It is recommended to configure it when using object storage for better performance.
-## A local file directory, defaults to `{data_home}/object_cache/read`. An empty string means disabling.
+## A local file directory, defaults to `{data_home}`. An empty string means disabling.
 ## @toml2docs:none-default
 #+ cache_path = ""
 
@@ -478,7 +478,7 @@ auto_flush_interval = "1h"
 ## Whether to enable the experimental write cache, it's enabled by default when using object storage. It is recommended to enable it when using object storage for better performance.
 enable_experimental_write_cache = false
 
-## File system path for write cache, defaults to `{data_home}/object_cache/write`.
+## File system path for write cache, defaults to `{data_home}`.
 experimental_write_cache_path = ""
 
 ## Capacity for write cache. If your disk space is sufficient, it is recommended to set it larger.
@@ -550,7 +550,7 @@ metadata_cache_size = "64MiB"
 content_cache_size = "128MiB"
 
 ## Page size for inverted index content cache.
-content_cache_page_size = "8MiB"
+content_cache_page_size = "64KiB"
 
 ## The options for full-text index in Mito engine.
 [region_engine.mito.fulltext_index]
@@ -576,6 +576,30 @@ apply_on_query = "auto"
 ## - `[size]` e.g. `64MB`: fixed memory threshold
 mem_threshold_on_create = "auto"
 
+## The options for bloom filter index in Mito engine.
+[region_engine.mito.bloom_filter_index]
+
+## Whether to create the index on flush.
+## - `auto`: automatically (default)
+## - `disable`: never
+create_on_flush = "auto"
+
+## Whether to create the index on compaction.
+## - `auto`: automatically (default)
+## - `disable`: never
+create_on_compaction = "auto"
+
+## Whether to apply the index on query
+## - `auto`: automatically (default)
+## - `disable`: never
+apply_on_query = "auto"
+
+## Memory threshold for the index creation.
+## - `auto`: automatically determine the threshold based on the system memory size (default)
+## - `unlimited`: no memory limit
+## - `[size]` e.g. `64MB`: fixed memory threshold
+mem_threshold_on_create = "auto"
+
 [region_engine.mito.memtable]
 ## Memtable type.
 ## - `time_series`: time-series memtable

config/frontend.example.toml

@@ -2,6 +2,10 @@
 ## @toml2docs:none-default
 default_timezone = "UTC"
 
+## The maximum in-flight write bytes.
+## @toml2docs:none-default
+#+ max_in_flight_write_bytes = "500MB"
+
 ## The runtime options.
 #+ [runtime]
 ## The number of threads to execute the runtime for global read operations.

config/standalone.example.toml

@@ -18,6 +18,10 @@ max_concurrent_queries = 0
 ## Enable telemetry to collect anonymous usage data. Enabled by default.
 #+ enable_telemetry = true
 
+## The maximum in-flight write bytes.
+## @toml2docs:none-default
+#+ max_in_flight_write_bytes = "500MB"
+
 ## The runtime options.
 #+ [runtime]
 ## The number of threads to execute the runtime for global read operations.
@@ -589,7 +593,7 @@ metadata_cache_size = "64MiB"
 content_cache_size = "128MiB"
 
 ## Page size for inverted index content cache.
-content_cache_page_size = "8MiB"
+content_cache_page_size = "64KiB"
 
 ## The options for full-text index in Mito engine.
 [region_engine.mito.fulltext_index]
@@ -615,6 +619,30 @@ apply_on_query = "auto"
 ## - `[size]` e.g. `64MB`: fixed memory threshold
 mem_threshold_on_create = "auto"
 
+## The options for bloom filter in Mito engine.
+[region_engine.mito.bloom_filter_index]
+
+## Whether to create the bloom filter on flush.
+## - `auto`: automatically (default)
+## - `disable`: never
+create_on_flush = "auto"
+
+## Whether to create the bloom filter on compaction.
+## - `auto`: automatically (default)
+## - `disable`: never
+create_on_compaction = "auto"
+
+## Whether to apply the bloom filter on query
+## - `auto`: automatically (default)
+## - `disable`: never
+apply_on_query = "auto"
+
+## Memory threshold for bloom filter creation.
+## - `auto`: automatically determine the threshold based on the system memory size (default)
+## - `unlimited`: no memory limit
+## - `[size]` e.g. `64MB`: fixed memory threshold
+mem_threshold_on_create = "auto"
+
 [region_engine.mito.memtable]
 ## Memtable type.
 ## - `time_series`: time-series memtable

cyborg/bin/bump-doc-version.ts

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2023 Greptime Team
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import * as core from "@actions/core";
+import {obtainClient} from "@/common";
+
+async function triggerWorkflow(workflowId: string, version: string) {
+  const docsClient = obtainClient("DOCS_REPO_TOKEN")
+  try {
+    await docsClient.rest.actions.createWorkflowDispatch({
+      owner: "GreptimeTeam",
+      repo: "docs",
+      workflow_id: workflowId,
+      ref: "main",
+      inputs: {
+        version,
+      },
+    });
+    console.log(`Successfully triggered ${workflowId} workflow with version ${version}`);
+  } catch (error) {
+    core.setFailed(`Failed to trigger workflow: ${error.message}`);
+  }
+}
+
+function determineWorkflow(version: string): [string, string] {
+  // Check if it's a nightly version
+  if (version.includes('nightly')) {
+    return ['bump-nightly-version.yml', version];
+  }
+
+  const parts = version.split('.');
+
+  if (parts.length !== 3) {
+    throw new Error('Invalid version format');
+  }
+
+  // If patch version (last number) is 0, it's a major version
+  // Return only major.minor version
+  if (parts[2] === '0') {
+    return ['bump-version.yml', `${parts[0]}.${parts[1]}`];
+  }
+
+  // Otherwise it's a patch version, use full version
+  return ['bump-patch-version.yml', version];
+}
+
+const version = process.env.VERSION;
+if (!version) {
+  core.setFailed("VERSION environment variable is required");
+  process.exit(1);
+}
+
+// Remove 'v' prefix if exists
+const cleanVersion = version.startsWith('v') ? version.slice(1) : version;
+
+try {
+  const [workflowId, apiVersion] = determineWorkflow(cleanVersion);
+  triggerWorkflow(workflowId, apiVersion);
+} catch (error) {
+  core.setFailed(`Error processing version: ${error.message}`);
+  process.exit(1);
+}

docker/buildx/centos/Dockerfile

@@ -13,8 +13,6 @@ RUN yum install -y epel-release  \
     openssl \
     openssl-devel  \
     centos-release-scl  \
-    rh-python38  \
-    rh-python38-python-devel \
     which
 
 # Install protoc
@@ -43,8 +41,6 @@ RUN yum install -y epel-release \
     openssl \
     openssl-devel  \
     centos-release-scl  \
-    rh-python38  \
-    rh-python38-python-devel \
     which
 
 WORKDIR /greptime

docker/buildx/ubuntu/Dockerfile

@@ -20,10 +20,7 @@ RUN --mount=type=cache,target=/var/cache/apt \
     curl \
     git \
     build-essential \
-    pkg-config \
-    python3.10 \
-    python3.10-dev \
-    python3-pip
+    pkg-config
 
 # Install Rust.
 SHELL ["/bin/bash", "-c"]
@@ -46,15 +43,8 @@ ARG OUTPUT_DIR
 
 RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get \
     -y install ca-certificates \
-    python3.10 \
-    python3.10-dev \
-    python3-pip \
     curl
 
-COPY ./docker/python/requirements.txt /etc/greptime/requirements.txt
-
-RUN python3 -m pip install -r /etc/greptime/requirements.txt
-
 WORKDIR /greptime
 COPY --from=builder /out/target/${OUTPUT_DIR}/greptime /greptime/bin/
 ENV PATH /greptime/bin/:$PATH

docker/ci/centos/Dockerfile

@@ -7,9 +7,7 @@ RUN sed -i s/^#.*baseurl=http/baseurl=http/g /etc/yum.repos.d/*.repo
 RUN yum install -y epel-release \
     openssl \
     openssl-devel  \
-    centos-release-scl  \
-    rh-python38  \
-    rh-python38-python-devel
+    centos-release-scl
 
 ARG TARGETARCH
 

docker/ci/ubuntu/Dockerfile

@@ -8,15 +8,8 @@ ARG TARGET_BIN=greptime
 
 RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
     ca-certificates \
-    python3.10 \
-    python3.10-dev \
-    python3-pip \
     curl
 
-COPY $DOCKER_BUILD_ROOT/docker/python/requirements.txt /etc/greptime/requirements.txt
-
-RUN python3 -m pip install -r /etc/greptime/requirements.txt
-
 ARG TARGETARCH
 
 ADD $TARGETARCH/$TARGET_BIN /greptime/bin/

docker/dev-builder/ubuntu/Dockerfile

@@ -15,8 +15,8 @@ RUN apt-get update && \
 RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
     libssl-dev \
     tzdata \
-    protobuf-compiler \
     curl \
+    unzip \
     ca-certificates \
     git \
     build-essential \
@@ -24,6 +24,20 @@ RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
     python3.10 \
     python3.10-dev
 
+ARG TARGETPLATFORM
+RUN echo "target platform: $TARGETPLATFORM"
+
+# 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/v29.1/protoc-29.1-linux-aarch_64.zip && \
+    unzip protoc-29.1-linux-aarch_64.zip -d protoc3; \
+elif [ "$TARGETPLATFORM" = "linux/amd64" ]; then \
+    curl -OL https://github.com/protocolbuffers/protobuf/releases/download/v29.1/protoc-29.1-linux-x86_64.zip && \
+    unzip protoc-29.1-linux-x86_64.zip -d protoc3; \
+fi
+RUN mv protoc3/bin/* /usr/local/bin/
+RUN mv protoc3/include/* /usr/local/include/
+
 # https://github.com/GreptimeTeam/greptimedb/actions/runs/10935485852/job/30357457188#step:3:7106
 # `aws-lc-sys` require gcc >= 10.3.0 to work, hence alias to use gcc-10
 RUN apt-get remove -y gcc-9 g++-9 cpp-9 && \
@@ -49,7 +63,7 @@ RUN apt-get -y purge python3.8 && \
 # 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 *
+RUN git config --global --add safe.directory '*'
 
 # Install Python dependencies.
 COPY $DOCKER_BUILD_ROOT/docker/python/requirements.txt /etc/greptime/requirements.txt

rust-toolchain.toml

@@ -1,3 +1,3 @@
 [toolchain]
 channel = "nightly-2024-10-19"
-components = ["rust-analyzer"]
+components = ["rust-analyzer", "llvm-tools"]

scripts/check-snafu.py

@@ -14,6 +14,7 @@
 
 import os
 import re
+from multiprocessing import Pool
 
 
 def find_rust_files(directory):
@@ -33,13 +34,11 @@ def extract_branch_names(file_content):
     return pattern.findall(file_content)
 
 
-def check_snafu_in_files(branch_name, rust_files):
+def check_snafu_in_files(branch_name, rust_files_content):
     branch_name_snafu = f"{branch_name}Snafu"
-    for rust_file in rust_files:
-        with open(rust_file, "r") as file:
-            content = file.read()
-            if branch_name_snafu in content:
-                return True
+    for content in rust_files_content.values():
+        if branch_name_snafu in content:
+            return True
     return False
 
 
@@ -49,21 +48,24 @@ def main():
 
     for error_file in error_files:
         with open(error_file, "r") as file:
-            content = file.read()
-            branch_names.extend(extract_branch_names(content))
+            branch_names.extend(extract_branch_names(file.read()))
 
-    unused_snafu = [
-        branch_name
-        for branch_name in branch_names
-        if not check_snafu_in_files(branch_name, other_rust_files)
-    ]
+    # Read all rust files into memory once
+    rust_files_content = {}
+    for rust_file in other_rust_files:
+        with open(rust_file, "r") as file:
+            rust_files_content[rust_file] = file.read()
+
+    with Pool() as pool:
+        results = pool.starmap(
+            check_snafu_in_files, [(bn, rust_files_content) for bn in branch_names]
+        )
+    unused_snafu = [bn for bn, found in zip(branch_names, results) if not found]
 
     if unused_snafu:
         print("Unused error variants:")
         for name in unused_snafu:
             print(name)
-
-    if unused_snafu:
         raise SystemExit(1)
 
 

shell.nix

@@ -1,5 +1,5 @@
 let
-  nixpkgs = fetchTarball "https://github.com/NixOS/nixpkgs/tarball/nixos-unstable";
+  nixpkgs = fetchTarball "https://github.com/NixOS/nixpkgs/tarball/nixos-24.11";
   fenix = import (fetchTarball "https://github.com/nix-community/fenix/archive/main.tar.gz") {};
   pkgs = import nixpkgs { config = {}; overlays = []; };
 in
@@ -11,16 +11,20 @@ pkgs.mkShell rec {
     clang
     gcc
     protobuf
+    gnumake
     mold
     (fenix.fromToolchainFile {
       dir = ./.;
     })
     cargo-nextest
+    cargo-llvm-cov
     taplo
+    curl
   ];
 
   buildInputs = with pkgs; [
     libgit2
+    libz
   ];
 
   LD_LIBRARY_PATH = pkgs.lib.makeLibraryPath buildInputs;

src/auth/src/permission.rs

@@ -25,6 +25,7 @@ pub enum PermissionReq<'a> {
     GrpcRequest(&'a Request),
     SqlStatement(&'a Statement),
     PromQuery,
+    LogQuery,
     Opentsdb,
     LineProtocol,
     PromStoreWrite,

src/catalog/src/error.rs

@@ -64,6 +64,13 @@ pub enum Error {
         source: BoxedError,
     },
 
+    #[snafu(display("Failed to list flow stats"))]
+    ListFlowStats {
+        #[snafu(implicit)]
+        location: Location,
+        source: BoxedError,
+    },
+
     #[snafu(display("Failed to list flows in catalog {catalog}"))]
     ListFlows {
         #[snafu(implicit)]
@@ -326,6 +333,7 @@ impl ErrorExt for Error {
             | Error::ListSchemas { source, .. }
             | Error::ListTables { source, .. }
             | Error::ListFlows { source, .. }
+            | Error::ListFlowStats { source, .. }
             | Error::ListProcedures { source, .. }
             | Error::ListRegionStats { source, .. }
             | Error::ConvertProtoData { source, .. } => source.status_code(),

src/catalog/src/information_extension.rs

@@ -17,6 +17,7 @@ use common_error::ext::BoxedError;
 use common_meta::cluster::{ClusterInfo, NodeInfo};
 use common_meta::datanode::RegionStat;
 use common_meta::ddl::{ExecutorContext, ProcedureExecutor};
+use common_meta::key::flow::flow_state::FlowStat;
 use common_meta::rpc::procedure;
 use common_procedure::{ProcedureInfo, ProcedureState};
 use meta_client::MetaClientRef;
@@ -89,4 +90,12 @@ impl InformationExtension for DistributedInformationExtension {
             .map_err(BoxedError::new)
             .context(error::ListRegionStatsSnafu)
     }
+
+    async fn flow_stats(&self) -> std::result::Result<Option<FlowStat>, Self::Error> {
+        self.meta_client
+            .list_flow_stats()
+            .await
+            .map_err(BoxedError::new)
+            .context(crate::error::ListFlowStatsSnafu)
+    }
 }

src/catalog/src/kvbackend/table_cache.rs

@@ -38,7 +38,7 @@ pub fn new_table_cache(
 ) -> TableCache {
     let init = init_factory(table_info_cache, table_name_cache);
 
-    CacheContainer::new(name, cache, Box::new(invalidator), init, Box::new(filter))
+    CacheContainer::new(name, cache, Box::new(invalidator), init, filter)
 }
 
 fn init_factory(

src/catalog/src/system_schema/information_schema.rs

@@ -35,6 +35,7 @@ use common_catalog::consts::{self, DEFAULT_CATALOG_NAME, INFORMATION_SCHEMA_NAME
 use common_error::ext::ErrorExt;
 use common_meta::cluster::NodeInfo;
 use common_meta::datanode::RegionStat;
+use common_meta::key::flow::flow_state::FlowStat;
 use common_meta::key::flow::FlowMetadataManager;
 use common_procedure::ProcedureInfo;
 use common_recordbatch::SendableRecordBatchStream;
@@ -192,6 +193,7 @@ impl SystemSchemaProviderInner for InformationSchemaProvider {
             )) as _),
             FLOWS => Some(Arc::new(InformationSchemaFlows::new(
                 self.catalog_name.clone(),
+                self.catalog_manager.clone(),
                 self.flow_metadata_manager.clone(),
             )) as _),
             PROCEDURE_INFO => Some(
@@ -338,6 +340,9 @@ pub trait InformationExtension {
 
     /// Gets the region statistics.
     async fn region_stats(&self) -> std::result::Result<Vec<RegionStat>, Self::Error>;
+
+    /// Get the flow statistics. If no flownode is available, return `None`.
+    async fn flow_stats(&self) -> std::result::Result<Option<FlowStat>, Self::Error>;
 }
 
 pub struct NoopInformationExtension;
@@ -357,4 +362,8 @@ impl InformationExtension for NoopInformationExtension {
     async fn region_stats(&self) -> std::result::Result<Vec<RegionStat>, Self::Error> {
         Ok(vec![])
     }
+
+    async fn flow_stats(&self) -> std::result::Result<Option<FlowStat>, Self::Error> {
+        Ok(None)
+    }
 }

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

@@ -12,11 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
+use std::sync::{Arc, Weak};
 
 use common_catalog::consts::INFORMATION_SCHEMA_FLOW_TABLE_ID;
 use common_error::ext::BoxedError;
 use common_meta::key::flow::flow_info::FlowInfoValue;
+use common_meta::key::flow::flow_state::FlowStat;
 use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::FlowId;
 use common_recordbatch::adapter::RecordBatchStreamAdapter;
@@ -28,7 +29,9 @@ use datatypes::prelude::ConcreteDataType as CDT;
 use datatypes::scalars::ScalarVectorBuilder;
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
 use datatypes::value::Value;
-use datatypes::vectors::{Int64VectorBuilder, StringVectorBuilder, UInt32VectorBuilder, VectorRef};
+use datatypes::vectors::{
+    Int64VectorBuilder, StringVectorBuilder, UInt32VectorBuilder, UInt64VectorBuilder, VectorRef,
+};
 use futures::TryStreamExt;
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::{ScanRequest, TableId};
@@ -38,6 +41,8 @@ use crate::error::{
 };
 use crate::information_schema::{Predicates, FLOWS};
 use crate::system_schema::information_schema::InformationTable;
+use crate::system_schema::utils;
+use crate::CatalogManager;
 
 const INIT_CAPACITY: usize = 42;
 
@@ -45,6 +50,7 @@ const INIT_CAPACITY: usize = 42;
 // pk is (flow_name, flow_id, table_catalog)
 pub const FLOW_NAME: &str = "flow_name";
 pub const FLOW_ID: &str = "flow_id";
+pub const STATE_SIZE: &str = "state_size";
 pub const TABLE_CATALOG: &str = "table_catalog";
 pub const FLOW_DEFINITION: &str = "flow_definition";
 pub const COMMENT: &str = "comment";
@@ -55,20 +61,24 @@ pub const FLOWNODE_IDS: &str = "flownode_ids";
 pub const OPTIONS: &str = "options";
 
 /// The `information_schema.flows` to provides information about flows in databases.
+///
 pub(super) struct InformationSchemaFlows {
     schema: SchemaRef,
     catalog_name: String,
+    catalog_manager: Weak<dyn CatalogManager>,
     flow_metadata_manager: Arc<FlowMetadataManager>,
 }
 
 impl InformationSchemaFlows {
     pub(super) fn new(
         catalog_name: String,
+        catalog_manager: Weak<dyn CatalogManager>,
         flow_metadata_manager: Arc<FlowMetadataManager>,
     ) -> Self {
         Self {
             schema: Self::schema(),
             catalog_name,
+            catalog_manager,
             flow_metadata_manager,
         }
     }
@@ -80,6 +90,7 @@ impl InformationSchemaFlows {
             vec![
                 (FLOW_NAME, CDT::string_datatype(), false),
                 (FLOW_ID, CDT::uint32_datatype(), false),
+                (STATE_SIZE, CDT::uint64_datatype(), true),
                 (TABLE_CATALOG, CDT::string_datatype(), false),
                 (FLOW_DEFINITION, CDT::string_datatype(), false),
                 (COMMENT, CDT::string_datatype(), true),
@@ -99,6 +110,7 @@ impl InformationSchemaFlows {
         InformationSchemaFlowsBuilder::new(
             self.schema.clone(),
             self.catalog_name.clone(),
+            self.catalog_manager.clone(),
             &self.flow_metadata_manager,
         )
     }
@@ -144,10 +156,12 @@ impl InformationTable for InformationSchemaFlows {
 struct InformationSchemaFlowsBuilder {
     schema: SchemaRef,
     catalog_name: String,
+    catalog_manager: Weak<dyn CatalogManager>,
     flow_metadata_manager: Arc<FlowMetadataManager>,
 
     flow_names: StringVectorBuilder,
     flow_ids: UInt32VectorBuilder,
+    state_sizes: UInt64VectorBuilder,
     table_catalogs: StringVectorBuilder,
     raw_sqls: StringVectorBuilder,
     comments: StringVectorBuilder,
@@ -162,15 +176,18 @@ impl InformationSchemaFlowsBuilder {
     fn new(
         schema: SchemaRef,
         catalog_name: String,
+        catalog_manager: Weak<dyn CatalogManager>,
         flow_metadata_manager: &Arc<FlowMetadataManager>,
     ) -> Self {
         Self {
             schema,
             catalog_name,
+            catalog_manager,
             flow_metadata_manager: flow_metadata_manager.clone(),
 
             flow_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             flow_ids: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
+            state_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
             table_catalogs: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             raw_sqls: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             comments: StringVectorBuilder::with_capacity(INIT_CAPACITY),
@@ -195,6 +212,11 @@ impl InformationSchemaFlowsBuilder {
             .flow_names(&catalog_name)
             .await;
 
+        let flow_stat = {
+            let information_extension = utils::information_extension(&self.catalog_manager)?;
+            information_extension.flow_stats().await?
+        };
+
         while let Some((flow_name, flow_id)) = stream
             .try_next()
             .await
@@ -213,7 +235,7 @@ impl InformationSchemaFlowsBuilder {
                     catalog_name: catalog_name.to_string(),
                     flow_name: flow_name.to_string(),
                 })?;
-            self.add_flow(&predicates, flow_id.flow_id(), flow_info)?;
+            self.add_flow(&predicates, flow_id.flow_id(), flow_info, &flow_stat)?;
         }
 
         self.finish()
@@ -224,6 +246,7 @@ impl InformationSchemaFlowsBuilder {
         predicates: &Predicates,
         flow_id: FlowId,
         flow_info: FlowInfoValue,
+        flow_stat: &Option<FlowStat>,
     ) -> Result<()> {
         let row = [
             (FLOW_NAME, &Value::from(flow_info.flow_name().to_string())),
@@ -238,6 +261,11 @@ impl InformationSchemaFlowsBuilder {
         }
         self.flow_names.push(Some(flow_info.flow_name()));
         self.flow_ids.push(Some(flow_id));
+        self.state_sizes.push(
+            flow_stat
+                .as_ref()
+                .and_then(|state| state.state_size.get(&flow_id).map(|v| *v as u64)),
+        );
         self.table_catalogs.push(Some(flow_info.catalog_name()));
         self.raw_sqls.push(Some(flow_info.raw_sql()));
         self.comments.push(Some(flow_info.comment()));
@@ -270,6 +298,7 @@ impl InformationSchemaFlowsBuilder {
         let columns: Vec<VectorRef> = vec![
             Arc::new(self.flow_names.finish()),
             Arc::new(self.flow_ids.finish()),
+            Arc::new(self.state_sizes.finish()),
             Arc::new(self.table_catalogs.finish()),
             Arc::new(self.raw_sqls.finish()),
             Arc::new(self.comments.finish()),

src/cli/Cargo.toml

@@ -15,7 +15,7 @@ cache.workspace = true
 catalog.workspace = true
 chrono.workspace = true
 clap.workspace = true
-client.workspace = true
+client = { workspace = true, features = ["testing"] }
 common-base.workspace = true
 common-catalog.workspace = true
 common-config.workspace = true
@@ -56,7 +56,6 @@ tokio.workspace = true
 tracing-appender.workspace = true
 
 [dev-dependencies]
-client = { workspace = true, features = ["testing"] }
 common-test-util.workspace = true
 common-version.workspace = true
 serde.workspace = true

src/cli/src/repl.rs

@@ -34,7 +34,7 @@ use common_query::Output;
 use common_recordbatch::RecordBatches;
 use common_telemetry::debug;
 use either::Either;
-use meta_client::client::MetaClientBuilder;
+use meta_client::client::{ClusterKvBackend, MetaClientBuilder};
 use query::datafusion::DatafusionQueryEngine;
 use query::parser::QueryLanguageParser;
 use query::query_engine::{DefaultSerializer, QueryEngineState};

src/cmd/src/standalone.rs

@@ -22,6 +22,7 @@ use catalog::information_schema::InformationExtension;
 use catalog::kvbackend::KvBackendCatalogManager;
 use clap::Parser;
 use client::api::v1::meta::RegionRole;
+use common_base::readable_size::ReadableSize;
 use common_base::Plugins;
 use common_catalog::consts::{MIN_USER_FLOW_ID, MIN_USER_TABLE_ID};
 use common_config::{metadata_store_dir, Configurable, KvBackendConfig};
@@ -34,6 +35,7 @@ use common_meta::ddl::flow_meta::{FlowMetadataAllocator, FlowMetadataAllocatorRe
 use common_meta::ddl::table_meta::{TableMetadataAllocator, TableMetadataAllocatorRef};
 use common_meta::ddl::{DdlContext, NoopRegionFailureDetectorControl, ProcedureExecutorRef};
 use common_meta::ddl_manager::DdlManager;
+use common_meta::key::flow::flow_state::FlowStat;
 use common_meta::key::flow::{FlowMetadataManager, FlowMetadataManagerRef};
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
@@ -70,7 +72,7 @@ use servers::http::HttpOptions;
 use servers::tls::{TlsMode, TlsOption};
 use servers::Mode;
 use snafu::ResultExt;
-use tokio::sync::broadcast;
+use tokio::sync::{broadcast, RwLock};
 use tracing_appender::non_blocking::WorkerGuard;
 
 use crate::error::{
@@ -151,6 +153,7 @@ pub struct StandaloneOptions {
     pub tracing: TracingOptions,
     pub init_regions_in_background: bool,
     pub init_regions_parallelism: usize,
+    pub max_in_flight_write_bytes: Option<ReadableSize>,
 }
 
 impl Default for StandaloneOptions {
@@ -180,6 +183,7 @@ impl Default for StandaloneOptions {
             tracing: TracingOptions::default(),
             init_regions_in_background: false,
             init_regions_parallelism: 16,
+            max_in_flight_write_bytes: None,
         }
     }
 }
@@ -217,6 +221,7 @@ impl StandaloneOptions {
             user_provider: cloned_opts.user_provider,
             // Handle the export metrics task run by standalone to frontend for execution
             export_metrics: cloned_opts.export_metrics,
+            max_in_flight_write_bytes: cloned_opts.max_in_flight_write_bytes,
             ..Default::default()
         }
     }
@@ -507,7 +512,7 @@ impl StartCommand {
             procedure_manager.clone(),
         ));
         let catalog_manager = KvBackendCatalogManager::new(
-            information_extension,
+            information_extension.clone(),
             kv_backend.clone(),
             layered_cache_registry.clone(),
             Some(procedure_manager.clone()),
@@ -532,6 +537,14 @@ impl StartCommand {
                 .context(OtherSnafu)?,
         );
 
+        // set the ref to query for the local flow state
+        {
+            let flow_worker_manager = flownode.flow_worker_manager();
+            information_extension
+                .set_flow_worker_manager(flow_worker_manager.clone())
+                .await;
+        }
+
         let node_manager = Arc::new(StandaloneDatanodeManager {
             region_server: datanode.region_server(),
             flow_server: flownode.flow_worker_manager(),
@@ -669,6 +682,7 @@ pub struct StandaloneInformationExtension {
     region_server: RegionServer,
     procedure_manager: ProcedureManagerRef,
     start_time_ms: u64,
+    flow_worker_manager: RwLock<Option<Arc<FlowWorkerManager>>>,
 }
 
 impl StandaloneInformationExtension {
@@ -677,8 +691,15 @@ impl StandaloneInformationExtension {
             region_server,
             procedure_manager,
             start_time_ms: common_time::util::current_time_millis() as u64,
+            flow_worker_manager: RwLock::new(None),
         }
     }
+
+    /// Set the flow worker manager for the standalone instance.
+    pub async fn set_flow_worker_manager(&self, flow_worker_manager: Arc<FlowWorkerManager>) {
+        let mut guard = self.flow_worker_manager.write().await;
+        *guard = Some(flow_worker_manager);
+    }
 }
 
 #[async_trait::async_trait]
@@ -750,6 +771,18 @@ impl InformationExtension for StandaloneInformationExtension {
             .collect::<Vec<_>>();
         Ok(stats)
     }
+
+    async fn flow_stats(&self) -> std::result::Result<Option<FlowStat>, Self::Error> {
+        Ok(Some(
+            self.flow_worker_manager
+                .read()
+                .await
+                .as_ref()
+                .unwrap()
+                .gen_state_report()
+                .await,
+        ))
+    }
 }
 
 #[cfg(test)]

src/common/datasource/tests/orc/write.py

@@ -35,10 +35,23 @@ data = {
     "bigint_other": [5, -5, 1, 5, 5],
     "utf8_increase": ["a", "bb", "ccc", "dddd", "eeeee"],
     "utf8_decrease": ["eeeee", "dddd", "ccc", "bb", "a"],
-    "timestamp_simple": [datetime.datetime(2023, 4, 1, 20, 15, 30, 2000), datetime.datetime.fromtimestamp(int('1629617204525777000')/1000000000), datetime.datetime(2023, 1, 1), datetime.datetime(2023, 2, 1), datetime.datetime(2023, 3, 1)],
-    "date_simple": [datetime.date(2023, 4, 1), datetime.date(2023, 3, 1), datetime.date(2023, 1, 1), datetime.date(2023, 2, 1), datetime.date(2023, 3, 1)]
+    "timestamp_simple": [
+        datetime.datetime(2023, 4, 1, 20, 15, 30, 2000),
+        datetime.datetime.fromtimestamp(int("1629617204525777000") / 1000000000),
+        datetime.datetime(2023, 1, 1),
+        datetime.datetime(2023, 2, 1),
+        datetime.datetime(2023, 3, 1),
+    ],
+    "date_simple": [
+        datetime.date(2023, 4, 1),
+        datetime.date(2023, 3, 1),
+        datetime.date(2023, 1, 1),
+        datetime.date(2023, 2, 1),
+        datetime.date(2023, 3, 1),
+    ],
 }
 
+
 def infer_schema(data):
     schema = "struct<"
     for key, value in data.items():
@@ -56,7 +69,7 @@ def infer_schema(data):
         elif key.startswith("date"):
             dt = "date"
         else:
-            print(key,value,dt)
+            print(key, value, dt)
             raise NotImplementedError
         if key.startswith("double"):
             dt = "double"
@@ -68,7 +81,6 @@ def infer_schema(data):
     return schema
 
 
-
 def _write(
     schema: str,
     data,

src/common/error/Cargo.toml

@@ -8,6 +8,7 @@ license.workspace = true
 workspace = true
 
 [dependencies]
+http.workspace = true
 snafu.workspace = true
 strum.workspace = true
 tonic.workspace = true

src/common/error/src/lib.rs

@@ -18,9 +18,30 @@ pub mod ext;
 pub mod mock;
 pub mod status_code;
 
+use http::{HeaderMap, HeaderValue};
 pub use snafu;
 
 // HACK - these headers are here for shared in gRPC services. For common HTTP headers,
 // please define in `src/servers/src/http/header.rs`.
 pub const GREPTIME_DB_HEADER_ERROR_CODE: &str = "x-greptime-err-code";
 pub const GREPTIME_DB_HEADER_ERROR_MSG: &str = "x-greptime-err-msg";
+
+/// Create a http header map from error code and message.
+/// using `GREPTIME_DB_HEADER_ERROR_CODE` and `GREPTIME_DB_HEADER_ERROR_MSG` as keys.
+pub fn from_err_code_msg_to_header(code: u32, msg: &str) -> HeaderMap {
+    let mut header = HeaderMap::new();
+
+    let msg = HeaderValue::from_str(msg).unwrap_or_else(|_| {
+        HeaderValue::from_bytes(
+            &msg.as_bytes()
+                .iter()
+                .flat_map(|b| std::ascii::escape_default(*b))
+                .collect::<Vec<u8>>(),
+        )
+        .expect("Already escaped string should be valid ascii")
+    });
+
+    header.insert(GREPTIME_DB_HEADER_ERROR_CODE, code.into());
+    header.insert(GREPTIME_DB_HEADER_ERROR_MSG, msg);
+    header
+}

src/common/function/Cargo.toml

@@ -33,7 +33,7 @@ geo-types = { version = "0.7", optional = true }
 geohash = { version = "0.13", optional = true }
 h3o = { version = "0.6", optional = true }
 jsonb.workspace = true
-nalgebra = "0.33"
+nalgebra.workspace = true
 num = "0.4"
 num-traits = "0.2"
 once_cell.workspace = true

src/common/function/src/lib.rs

@@ -26,3 +26,4 @@ pub mod function_registry;
 pub mod handlers;
 pub mod helper;
 pub mod state;
+pub mod utils;

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

@@ -32,6 +32,7 @@ pub use scipy_stats_norm_cdf::ScipyStatsNormCdfAccumulatorCreator;
 pub use scipy_stats_norm_pdf::ScipyStatsNormPdfAccumulatorCreator;
 
 use crate::function_registry::FunctionRegistry;
+use crate::scalars::vector::sum::VectorSumCreator;
 
 /// A function creates `AggregateFunctionCreator`.
 /// "Aggregator" *is* AggregatorFunction. Since the later one is long, we named an short alias for it.
@@ -91,6 +92,7 @@ impl AggregateFunctions {
         register_aggr_func!("argmin", 1, ArgminAccumulatorCreator);
         register_aggr_func!("scipystatsnormcdf", 2, ScipyStatsNormCdfAccumulatorCreator);
         register_aggr_func!("scipystatsnormpdf", 2, ScipyStatsNormPdfAccumulatorCreator);
+        register_aggr_func!("vec_sum", 1, VectorSumCreator);
 
         #[cfg(feature = "geo")]
         register_aggr_func!(

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

@@ -204,20 +204,10 @@ impl PatternAst {
     fn convert_literal(column: &str, pattern: &str) -> Expr {
         logical_expr::col(column).like(logical_expr::lit(format!(
             "%{}%",
-            Self::escape_pattern(pattern)
+            crate::utils::escape_like_pattern(pattern)
         )))
     }
 
-    fn escape_pattern(pattern: &str) -> String {
-        pattern
-            .chars()
-            .flat_map(|c| match c {
-                '\\' | '%' | '_' => vec!['\\', c],
-                _ => vec![c],
-            })
-            .collect::<String>()
-    }
-
     /// Transform this AST with preset rules to make it correct.
     fn transform_ast(self) -> Result<Self> {
         self.transform_up(Self::collapse_binary_branch_fn)
@@ -735,7 +725,8 @@ struct Tokenizer {
 impl Tokenizer {
     pub fn tokenize(mut self, pattern: &str) -> Result<Vec<Token>> {
         let mut tokens = vec![];
-        while self.cursor < pattern.len() {
+        let char_len = pattern.chars().count();
+        while self.cursor < char_len {
             // TODO: collect pattern into Vec<char> if this tokenizer is bottleneck in the future
             let c = pattern.chars().nth(self.cursor).unwrap();
             match c {
@@ -804,7 +795,8 @@ impl Tokenizer {
         let mut phase = String::new();
         let mut is_quote_present = false;
 
-        while self.cursor < pattern.len() {
+        let char_len = pattern.chars().count();
+        while self.cursor < char_len {
             let mut c = pattern.chars().nth(self.cursor).unwrap();
 
             match c {
@@ -909,6 +901,26 @@ mod test {
                     Phase("c".to_string()),
                 ],
             ),
+            (
+                r#"中文 测试"#,
+                vec![Phase("中文".to_string()), Phase("测试".to_string())],
+            ),
+            (
+                r#"中文 AND 测试"#,
+                vec![Phase("中文".to_string()), And, Phase("测试".to_string())],
+            ),
+            (
+                r#"中文 +测试"#,
+                vec![Phase("中文".to_string()), Must, Phase("测试".to_string())],
+            ),
+            (
+                r#"中文 -测试"#,
+                vec![
+                    Phase("中文".to_string()),
+                    Negative,
+                    Phase("测试".to_string()),
+                ],
+            ),
         ];
 
         for (query, expected) in cases {
@@ -1040,6 +1052,61 @@ mod test {
                     ],
                 },
             ),
+            (
+                r#"中文 测试"#,
+                PatternAst::Binary {
+                    op: BinaryOp::Or,
+                    children: vec![
+                        PatternAst::Literal {
+                            op: UnaryOp::Optional,
+                            pattern: "中文".to_string(),
+                        },
+                        PatternAst::Literal {
+                            op: UnaryOp::Optional,
+                            pattern: "测试".to_string(),
+                        },
+                    ],
+                },
+            ),
+            (
+                r#"中文 AND 测试"#,
+                PatternAst::Binary {
+                    op: BinaryOp::And,
+                    children: vec![
+                        PatternAst::Literal {
+                            op: UnaryOp::Optional,
+                            pattern: "中文".to_string(),
+                        },
+                        PatternAst::Literal {
+                            op: UnaryOp::Optional,
+                            pattern: "测试".to_string(),
+                        },
+                    ],
+                },
+            ),
+            (
+                r#"中文 +测试"#,
+                PatternAst::Literal {
+                    op: UnaryOp::Must,
+                    pattern: "测试".to_string(),
+                },
+            ),
+            (
+                r#"中文 -测试"#,
+                PatternAst::Binary {
+                    op: BinaryOp::And,
+                    children: vec![
+                        PatternAst::Literal {
+                            op: UnaryOp::Negative,
+                            pattern: "测试".to_string(),
+                        },
+                        PatternAst::Literal {
+                            op: UnaryOp::Optional,
+                            pattern: "中文".to_string(),
+                        },
+                    ],
+                },
+            ),
         ];
 
         for (query, expected) in cases {

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

@@ -14,9 +14,14 @@
 
 mod convert;
 mod distance;
-pub(crate) mod impl_conv;
+mod elem_sum;
+pub mod impl_conv;
 mod scalar_add;
 mod scalar_mul;
+mod sub;
+pub(crate) mod sum;
+mod vector_div;
+mod vector_mul;
 
 use std::sync::Arc;
 
@@ -38,5 +43,11 @@ impl VectorFunction {
         // scalar calculation
         registry.register(Arc::new(scalar_add::ScalarAddFunction));
         registry.register(Arc::new(scalar_mul::ScalarMulFunction));
+
+        // vector calculation
+        registry.register(Arc::new(vector_mul::VectorMulFunction));
+        registry.register(Arc::new(vector_div::VectorDivFunction));
+        registry.register(Arc::new(sub::SubFunction));
+        registry.register(Arc::new(elem_sum::ElemSumFunction));
     }
 }

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

@@ -0,0 +1,129 @@
+// 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::{Float32VectorBuilder, MutableVector, VectorRef};
+use nalgebra::DVectorView;
+use snafu::ensure;
+
+use crate::function::{Function, FunctionContext};
+use crate::scalars::vector::impl_conv::{as_veclit, as_veclit_if_const};
+
+const NAME: &str = "vec_elem_sum";
+
+#[derive(Debug, Clone, Default)]
+pub struct ElemSumFunction;
+
+impl Function for ElemSumFunction {
+    fn name(&self) -> &str {
+        NAME
+    }
+
+    fn return_type(
+        &self,
+        _input_types: &[ConcreteDataType],
+    ) -> common_query::error::Result<ConcreteDataType> {
+        Ok(ConcreteDataType::float32_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 = Float32VectorBuilder::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(DVectorView::from_slice(&arg0, arg0.len()).sum()));
+        }
+
+        Ok(result.to_vector())
+    }
+}
+
+impl Display for ElemSumFunction {
+    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 datatypes::vectors::StringVector;
+
+    use super::*;
+    use crate::function::FunctionContext;
+
+    #[test]
+    fn test_elem_sum() {
+        let func = ElemSumFunction;
+
+        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,
+        ]));
+
+        let result = func.eval(FunctionContext::default(), &[input0]).unwrap();
+
+        let result = result.as_ref();
+        assert_eq!(result.len(), 3);
+        assert_eq!(result.get_ref(0).as_f32().unwrap(), Some(6.0));
+        assert_eq!(result.get_ref(1).as_f32().unwrap(), Some(15.0));
+        assert_eq!(result.get_ref(2).as_f32().unwrap(), None);
+    }
+}

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

@@ -0,0 +1,223 @@
+// 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;
+use datatypes::prelude::ConcreteDataType;
+use datatypes::scalars::ScalarVectorBuilder;
+use datatypes::vectors::{BinaryVectorBuilder, MutableVector, VectorRef};
+use nalgebra::DVectorView;
+use snafu::ensure;
+
+use crate::function::{Function, FunctionContext};
+use crate::helper;
+use crate::scalars::vector::impl_conv::{as_veclit, as_veclit_if_const, veclit_to_binlit};
+
+const NAME: &str = "vec_sub";
+
+/// Subtracts corresponding elements of two vectors, returns a vector.
+///
+/// # Example
+///
+/// ```sql
+/// SELECT vec_to_string(vec_sub("[1.0, 1.0]", "[1.0, 2.0]")) as result;
+///
+/// +---------------------------------------------------------------+
+/// | vec_to_string(vec_sub(Utf8("[1.0, 1.0]"),Utf8("[1.0, 2.0]"))) |
+/// +---------------------------------------------------------------+
+/// | [0,-1]                                                        |
+/// +---------------------------------------------------------------+
+///
+/// -- Negative scalar to simulate subtraction
+/// SELECT vec_to_string(vec_sub('[-1.0, -1.0]', '[1.0, 2.0]'));
+///
+/// +-----------------------------------------------------------------+
+/// | vec_to_string(vec_sub(Utf8("[-1.0, -1.0]"),Utf8("[1.0, 2.0]"))) |
+/// +-----------------------------------------------------------------+
+/// | [-2,-3]                                                         |
+/// +-----------------------------------------------------------------+
+///
+#[derive(Debug, Clone, Default)]
+pub struct SubFunction;
+
+impl Function for SubFunction {
+    fn name(&self) -> &str {
+        NAME
+    }
+
+    fn return_type(
+        &self,
+        _input_types: &[ConcreteDataType],
+    ) -> common_query::error::Result<ConcreteDataType> {
+        Ok(ConcreteDataType::binary_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        helper::one_of_sigs2(
+            vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::binary_datatype(),
+            ],
+            vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::binary_datatype(),
+            ],
+        )
+    }
+
+    fn eval(
+        &self,
+        _func_ctx: FunctionContext,
+        columns: &[VectorRef],
+    ) -> common_query::error::Result<VectorRef> {
+        ensure!(
+            columns.len() == 2,
+            InvalidFuncArgsSnafu {
+                err_msg: format!(
+                    "The length of the args is not correct, expect exactly two, have: {}",
+                    columns.len()
+                )
+            }
+        );
+        let arg0 = &columns[0];
+        let arg1 = &columns[1];
+
+        ensure!(
+            arg0.len() == arg1.len(),
+            InvalidFuncArgsSnafu {
+                err_msg: format!(
+                    "The lengths of the vector are not aligned, args 0: {}, args 1: {}",
+                    arg0.len(),
+                    arg1.len(),
+                )
+            }
+        );
+
+        let len = arg0.len();
+        let mut result = BinaryVectorBuilder::with_capacity(len);
+        if len == 0 {
+            return Ok(result.to_vector());
+        }
+
+        let arg0_const = as_veclit_if_const(arg0)?;
+        let arg1_const = as_veclit_if_const(arg1)?;
+
+        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 arg1 = match arg1_const.as_ref() {
+                Some(arg1) => Some(Cow::Borrowed(arg1.as_ref())),
+                None => as_veclit(arg1.get_ref(i))?,
+            };
+            let (Some(arg0), Some(arg1)) = (arg0, arg1) else {
+                result.push_null();
+                continue;
+            };
+            let vec0 = DVectorView::from_slice(&arg0, arg0.len());
+            let vec1 = DVectorView::from_slice(&arg1, arg1.len());
+
+            let vec_res = vec0 - vec1;
+            let veclit = vec_res.as_slice();
+            let binlit = veclit_to_binlit(veclit);
+            result.push(Some(&binlit));
+        }
+
+        Ok(result.to_vector())
+    }
+}
+
+impl Display for SubFunction {
+    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_sub() {
+        let func = SubFunction;
+
+        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()),
+            None,
+        ]));
+
+        let result = func
+            .eval(FunctionContext::default(), &[input0, input1])
+            .unwrap();
+
+        let result = result.as_ref();
+        assert_eq!(result.len(), 4);
+        assert_eq!(
+            result.get_ref(0).as_binary().unwrap(),
+            Some(veclit_to_binlit(&[0.0, 1.0, 2.0]).as_slice())
+        );
+        assert_eq!(
+            result.get_ref(1).as_binary().unwrap(),
+            Some(veclit_to_binlit(&[-2.0, 0.0, 2.0]).as_slice())
+        );
+        assert!(result.get_ref(2).is_null());
+        assert!(result.get_ref(3).is_null());
+    }
+
+    #[test]
+    fn test_sub_error() {
+        let func = SubFunction;
+
+        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 lengths of the vector are not aligned, args 0: 4, args 1: 3"
+                )
+            }
+            _ => unreachable!(),
+        }
+    }
+}

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

@@ -0,0 +1,202 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use common_macro::{as_aggr_func_creator, AggrFuncTypeStore};
+use common_query::error::{CreateAccumulatorSnafu, Error, InvalidFuncArgsSnafu};
+use common_query::logical_plan::{Accumulator, AggregateFunctionCreator};
+use common_query::prelude::AccumulatorCreatorFunction;
+use datatypes::prelude::{ConcreteDataType, Value, *};
+use datatypes::vectors::VectorRef;
+use nalgebra::{Const, DVectorView, Dyn, OVector};
+use snafu::ensure;
+
+use crate::scalars::vector::impl_conv::{as_veclit, as_veclit_if_const, veclit_to_binlit};
+
+#[derive(Debug, Default)]
+pub struct VectorSum {
+    sum: Option<OVector<f32, Dyn>>,
+    has_null: bool,
+}
+
+#[as_aggr_func_creator]
+#[derive(Debug, Default, AggrFuncTypeStore)]
+pub struct VectorSumCreator {}
+
+impl AggregateFunctionCreator for VectorSumCreator {
+    fn creator(&self) -> AccumulatorCreatorFunction {
+        let creator: AccumulatorCreatorFunction = Arc::new(move |types: &[ConcreteDataType]| {
+            ensure!(
+                types.len() == 1,
+                InvalidFuncArgsSnafu {
+                    err_msg: format!(
+                        "The length of the args is not correct, expect exactly one, have: {}",
+                        types.len()
+                    )
+                }
+            );
+            let input_type = &types[0];
+            match input_type {
+                ConcreteDataType::String(_) | ConcreteDataType::Binary(_) => {
+                    Ok(Box::new(VectorSum::default()))
+                }
+                _ => {
+                    let err_msg = format!(
+                        "\"VEC_SUM\" aggregate function not support data type {:?}",
+                        input_type.logical_type_id(),
+                    );
+                    CreateAccumulatorSnafu { err_msg }.fail()?
+                }
+            }
+        });
+        creator
+    }
+
+    fn output_type(&self) -> common_query::error::Result<ConcreteDataType> {
+        Ok(ConcreteDataType::binary_datatype())
+    }
+
+    fn state_types(&self) -> common_query::error::Result<Vec<ConcreteDataType>> {
+        Ok(vec![self.output_type()?])
+    }
+}
+
+impl VectorSum {
+    fn inner(&mut self, len: usize) -> &mut OVector<f32, Dyn> {
+        self.sum
+            .get_or_insert_with(|| OVector::zeros_generic(Dyn(len), Const::<1>))
+    }
+
+    fn update(&mut self, values: &[VectorRef], is_update: bool) -> Result<(), Error> {
+        if values.is_empty() || self.has_null {
+            return Ok(());
+        };
+        let column = &values[0];
+        let len = column.len();
+
+        match as_veclit_if_const(column)? {
+            Some(column) => {
+                let vec_column = DVectorView::from_slice(&column, column.len()).scale(len as f32);
+                *self.inner(vec_column.len()) += vec_column;
+            }
+            None => {
+                for i in 0..len {
+                    let Some(arg0) = as_veclit(column.get_ref(i))? else {
+                        if is_update {
+                            self.has_null = true;
+                            self.sum = None;
+                        }
+                        return Ok(());
+                    };
+                    let vec_column = DVectorView::from_slice(&arg0, arg0.len());
+                    *self.inner(vec_column.len()) += vec_column;
+                }
+            }
+        }
+        Ok(())
+    }
+}
+
+impl Accumulator for VectorSum {
+    fn state(&self) -> common_query::error::Result<Vec<Value>> {
+        self.evaluate().map(|v| vec![v])
+    }
+
+    fn update_batch(&mut self, values: &[VectorRef]) -> common_query::error::Result<()> {
+        self.update(values, true)
+    }
+
+    fn merge_batch(&mut self, states: &[VectorRef]) -> common_query::error::Result<()> {
+        self.update(states, false)
+    }
+
+    fn evaluate(&self) -> common_query::error::Result<Value> {
+        match &self.sum {
+            None => Ok(Value::Null),
+            Some(vector) => Ok(Value::from(veclit_to_binlit(vector.as_slice()))),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use datatypes::vectors::{ConstantVector, StringVector};
+
+    use super::*;
+
+    #[test]
+    fn test_update_batch() {
+        // test update empty batch, expect not updating anything
+        let mut vec_sum = VectorSum::default();
+        vec_sum.update_batch(&[]).unwrap();
+        assert!(vec_sum.sum.is_none());
+        assert!(!vec_sum.has_null);
+        assert_eq!(Value::Null, vec_sum.evaluate().unwrap());
+
+        // test update one not-null value
+        let mut vec_sum = VectorSum::default();
+        let v: Vec<VectorRef> = vec![Arc::new(StringVector::from(vec![Some(
+            "[1.0,2.0,3.0]".to_string(),
+        )]))];
+        vec_sum.update_batch(&v).unwrap();
+        assert_eq!(
+            Value::from(veclit_to_binlit(&[1.0, 2.0, 3.0])),
+            vec_sum.evaluate().unwrap()
+        );
+
+        // test update one null value
+        let mut vec_sum = VectorSum::default();
+        let v: Vec<VectorRef> = vec![Arc::new(StringVector::from(vec![Option::<String>::None]))];
+        vec_sum.update_batch(&v).unwrap();
+        assert_eq!(Value::Null, vec_sum.evaluate().unwrap());
+
+        // test update no null-value batch
+        let mut vec_sum = VectorSum::default();
+        let v: Vec<VectorRef> = vec![Arc::new(StringVector::from(vec![
+            Some("[1.0,2.0,3.0]".to_string()),
+            Some("[4.0,5.0,6.0]".to_string()),
+            Some("[7.0,8.0,9.0]".to_string()),
+        ]))];
+        vec_sum.update_batch(&v).unwrap();
+        assert_eq!(
+            Value::from(veclit_to_binlit(&[12.0, 15.0, 18.0])),
+            vec_sum.evaluate().unwrap()
+        );
+
+        // test update null-value batch
+        let mut vec_sum = VectorSum::default();
+        let v: Vec<VectorRef> = vec![Arc::new(StringVector::from(vec![
+            Some("[1.0,2.0,3.0]".to_string()),
+            None,
+            Some("[7.0,8.0,9.0]".to_string()),
+        ]))];
+        vec_sum.update_batch(&v).unwrap();
+        assert_eq!(Value::Null, vec_sum.evaluate().unwrap());
+
+        // test update with constant vector
+        let mut vec_sum = VectorSum::default();
+        let v: Vec<VectorRef> = vec![Arc::new(ConstantVector::new(
+            Arc::new(StringVector::from_vec(vec!["[1.0,2.0,3.0]".to_string()])),
+            4,
+        ))];
+        vec_sum.update_batch(&v).unwrap();
+        assert_eq!(
+            Value::from(veclit_to_binlit(&[4.0, 8.0, 12.0])),
+            vec_sum.evaluate().unwrap()
+        );
+    }
+}

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

@@ -0,0 +1,218 @@
+// 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, Result};
+use common_query::prelude::Signature;
+use datatypes::prelude::ConcreteDataType;
+use datatypes::scalars::ScalarVectorBuilder;
+use datatypes::vectors::{BinaryVectorBuilder, MutableVector, VectorRef};
+use nalgebra::DVectorView;
+use snafu::ensure;
+
+use crate::function::{Function, FunctionContext};
+use crate::helper;
+use crate::scalars::vector::impl_conv::{as_veclit, as_veclit_if_const, veclit_to_binlit};
+
+const NAME: &str = "vec_div";
+
+/// Divides corresponding elements of two vectors.
+///
+/// # Example
+///
+/// ```sql
+/// SELECT vec_to_string(vec_div("[2, 4, 6]", "[2, 2, 2]")) as result;
+///
+/// +---------+
+/// | result  |
+/// +---------+
+/// | [1,2,3] |
+/// +---------+
+///
+/// ```
+#[derive(Debug, Clone, Default)]
+pub struct VectorDivFunction;
+
+impl Function for VectorDivFunction {
+    fn name(&self) -> &str {
+        NAME
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::binary_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        helper::one_of_sigs2(
+            vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::binary_datatype(),
+            ],
+            vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::binary_datatype(),
+            ],
+        )
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure!(
+            columns.len() == 2,
+            InvalidFuncArgsSnafu {
+                err_msg: format!(
+                    "The length of the args is not correct, expect exactly two, have: {}",
+                    columns.len()
+                ),
+            }
+        );
+
+        let arg0 = &columns[0];
+        let arg1 = &columns[1];
+
+        let len = arg0.len();
+        let mut result = BinaryVectorBuilder::with_capacity(len);
+        if len == 0 {
+            return Ok(result.to_vector());
+        }
+
+        let arg0_const = as_veclit_if_const(arg0)?;
+        let arg1_const = as_veclit_if_const(arg1)?;
+
+        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 arg1 = match arg1_const.as_ref() {
+                Some(arg1) => Some(Cow::Borrowed(arg1.as_ref())),
+                None => as_veclit(arg1.get_ref(i))?,
+            };
+
+            if let (Some(arg0), Some(arg1)) = (arg0, arg1) {
+                ensure!(
+                    arg0.len() == arg1.len(),
+                    InvalidFuncArgsSnafu {
+                        err_msg: format!(
+                            "The length of the vectors must match for division, have: {} vs {}",
+                            arg0.len(),
+                            arg1.len()
+                        ),
+                    }
+                );
+                let vec0 = DVectorView::from_slice(&arg0, arg0.len());
+                let vec1 = DVectorView::from_slice(&arg1, arg1.len());
+                let vec_res = vec0.component_div(&vec1);
+
+                let veclit = vec_res.as_slice();
+                let binlit = veclit_to_binlit(veclit);
+                result.push(Some(&binlit));
+            } else {
+                result.push_null();
+            }
+        }
+
+        Ok(result.to_vector())
+    }
+}
+
+impl Display for VectorDivFunction {
+    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;
+    use datatypes::vectors::StringVector;
+
+    use super::*;
+
+    #[test]
+    fn test_vector_mul() {
+        let func = VectorDivFunction;
+
+        let vec0 = vec![1.0, 2.0, 3.0];
+        let vec1 = vec![1.0, 1.0];
+        let (len0, len1) = (vec0.len(), vec1.len());
+        let input0 = Arc::new(StringVector::from(vec![Some(format!("{vec0:?}"))]));
+        let input1 = Arc::new(StringVector::from(vec![Some(format!("{vec1:?}"))]));
+
+        let err = func
+            .eval(FunctionContext::default(), &[input0, input1])
+            .unwrap_err();
+
+        match err {
+            error::Error::InvalidFuncArgs { err_msg, .. } => {
+                assert_eq!(
+                    err_msg,
+                    format!(
+                        "The length of the vectors must match for division, have: {} vs {}",
+                        len0, len1
+                    )
+                )
+            }
+            _ => unreachable!(),
+        }
+
+        let input0 = Arc::new(StringVector::from(vec![
+            Some("[1.0,2.0,3.0]".to_string()),
+            Some("[8.0,10.0,12.0]".to_string()),
+            Some("[7.0,8.0,9.0]".to_string()),
+            None,
+        ]));
+
+        let input1 = Arc::new(StringVector::from(vec![
+            Some("[1.0,1.0,1.0]".to_string()),
+            Some("[2.0,2.0,2.0]".to_string()),
+            None,
+            Some("[3.0,3.0,3.0]".to_string()),
+        ]));
+
+        let result = func
+            .eval(FunctionContext::default(), &[input0, input1])
+            .unwrap();
+
+        let result = result.as_ref();
+        assert_eq!(result.len(), 4);
+        assert_eq!(
+            result.get_ref(0).as_binary().unwrap(),
+            Some(veclit_to_binlit(&[1.0, 2.0, 3.0]).as_slice())
+        );
+        assert_eq!(
+            result.get_ref(1).as_binary().unwrap(),
+            Some(veclit_to_binlit(&[4.0, 5.0, 6.0]).as_slice())
+        );
+        assert!(result.get_ref(2).is_null());
+        assert!(result.get_ref(3).is_null());
+
+        let input0 = Arc::new(StringVector::from(vec![Some("[1.0,-2.0]".to_string())]));
+        let input1 = Arc::new(StringVector::from(vec![Some("[0.0,0.0]".to_string())]));
+
+        let result = func
+            .eval(FunctionContext::default(), &[input0, input1])
+            .unwrap();
+
+        let result = result.as_ref();
+        assert_eq!(
+            result.get_ref(0).as_binary().unwrap(),
+            Some(veclit_to_binlit(&[f64::INFINITY as f32, f64::NEG_INFINITY as f32]).as_slice())
+        );
+    }
+}

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

@@ -0,0 +1,205 @@
+// 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, Result};
+use common_query::prelude::Signature;
+use datatypes::prelude::ConcreteDataType;
+use datatypes::scalars::ScalarVectorBuilder;
+use datatypes::vectors::{BinaryVectorBuilder, MutableVector, VectorRef};
+use nalgebra::DVectorView;
+use snafu::ensure;
+
+use crate::function::{Function, FunctionContext};
+use crate::helper;
+use crate::scalars::vector::impl_conv::{as_veclit, as_veclit_if_const, veclit_to_binlit};
+
+const NAME: &str = "vec_mul";
+
+/// Multiplies corresponding elements of two vectors.
+///
+/// # Example
+///
+/// ```sql
+/// SELECT vec_to_string(vec_mul("[1, 2, 3]", "[1, 2, 3]")) as result;
+///
+/// +---------+
+/// | result  |
+/// +---------+
+/// | [1,4,9] |
+/// +---------+
+///
+/// ```
+#[derive(Debug, Clone, Default)]
+pub struct VectorMulFunction;
+
+impl Function for VectorMulFunction {
+    fn name(&self) -> &str {
+        NAME
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::binary_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        helper::one_of_sigs2(
+            vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::binary_datatype(),
+            ],
+            vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::binary_datatype(),
+            ],
+        )
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure!(
+            columns.len() == 2,
+            InvalidFuncArgsSnafu {
+                err_msg: format!(
+                    "The length of the args is not correct, expect exactly two, have: {}",
+                    columns.len()
+                ),
+            }
+        );
+
+        let arg0 = &columns[0];
+        let arg1 = &columns[1];
+
+        let len = arg0.len();
+        let mut result = BinaryVectorBuilder::with_capacity(len);
+        if len == 0 {
+            return Ok(result.to_vector());
+        }
+
+        let arg0_const = as_veclit_if_const(arg0)?;
+        let arg1_const = as_veclit_if_const(arg1)?;
+
+        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 arg1 = match arg1_const.as_ref() {
+                Some(arg1) => Some(Cow::Borrowed(arg1.as_ref())),
+                None => as_veclit(arg1.get_ref(i))?,
+            };
+
+            if let (Some(arg0), Some(arg1)) = (arg0, arg1) {
+                ensure!(
+                    arg0.len() == arg1.len(),
+                    InvalidFuncArgsSnafu {
+                        err_msg: format!(
+                            "The length of the vectors must match for multiplying, have: {} vs {}",
+                            arg0.len(),
+                            arg1.len()
+                        ),
+                    }
+                );
+                let vec0 = DVectorView::from_slice(&arg0, arg0.len());
+                let vec1 = DVectorView::from_slice(&arg1, arg1.len());
+                let vec_res = vec1.component_mul(&vec0);
+
+                let veclit = vec_res.as_slice();
+                let binlit = veclit_to_binlit(veclit);
+                result.push(Some(&binlit));
+            } else {
+                result.push_null();
+            }
+        }
+
+        Ok(result.to_vector())
+    }
+}
+
+impl Display for VectorMulFunction {
+    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;
+    use datatypes::vectors::StringVector;
+
+    use super::*;
+
+    #[test]
+    fn test_vector_mul() {
+        let func = VectorMulFunction;
+
+        let vec0 = vec![1.0, 2.0, 3.0];
+        let vec1 = vec![1.0, 1.0];
+        let (len0, len1) = (vec0.len(), vec1.len());
+        let input0 = Arc::new(StringVector::from(vec![Some(format!("{vec0:?}"))]));
+        let input1 = Arc::new(StringVector::from(vec![Some(format!("{vec1:?}"))]));
+
+        let err = func
+            .eval(FunctionContext::default(), &[input0, input1])
+            .unwrap_err();
+
+        match err {
+            error::Error::InvalidFuncArgs { err_msg, .. } => {
+                assert_eq!(
+                    err_msg,
+                    format!(
+                        "The length of the vectors must match for multiplying, have: {} vs {}",
+                        len0, len1
+                    )
+                )
+            }
+            _ => unreachable!(),
+        }
+
+        let input0 = Arc::new(StringVector::from(vec![
+            Some("[1.0,2.0,3.0]".to_string()),
+            Some("[8.0,10.0,12.0]".to_string()),
+            Some("[7.0,8.0,9.0]".to_string()),
+            None,
+        ]));
+
+        let input1 = Arc::new(StringVector::from(vec![
+            Some("[1.0,1.0,1.0]".to_string()),
+            Some("[2.0,2.0,2.0]".to_string()),
+            None,
+            Some("[3.0,3.0,3.0]".to_string()),
+        ]));
+
+        let result = func
+            .eval(FunctionContext::default(), &[input0, input1])
+            .unwrap();
+
+        let result = result.as_ref();
+        assert_eq!(result.len(), 4);
+        assert_eq!(
+            result.get_ref(0).as_binary().unwrap(),
+            Some(veclit_to_binlit(&[1.0, 2.0, 3.0]).as_slice())
+        );
+        assert_eq!(
+            result.get_ref(1).as_binary().unwrap(),
+            Some(veclit_to_binlit(&[16.0, 20.0, 24.0]).as_slice())
+        );
+        assert!(result.get_ref(2).is_null());
+        assert!(result.get_ref(3).is_null());
+    }
+}

src/common/function/src/utils.rs

@@ -0,0 +1,58 @@
+// 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.
+
+/// Escapes special characters in the provided pattern string for `LIKE`.
+///
+/// Specifically, it prefixes the backslash (`\`), percent (`%`), and underscore (`_`)
+/// characters with an additional backslash to ensure they are treated literally.
+///
+/// # Examples
+///
+/// ```rust
+/// let escaped = escape_pattern("100%_some\\path");
+/// assert_eq!(escaped, "100\\%\\_some\\\\path");
+/// ```
+pub fn escape_like_pattern(pattern: &str) -> String {
+    pattern
+        .chars()
+        .flat_map(|c| match c {
+            '\\' | '%' | '_' => vec!['\\', c],
+            _ => vec![c],
+        })
+        .collect::<String>()
+}
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_escape_like_pattern() {
+        assert_eq!(
+            escape_like_pattern("100%_some\\path"),
+            "100\\%\\_some\\\\path"
+        );
+        assert_eq!(escape_like_pattern(""), "");
+        assert_eq!(escape_like_pattern("hello"), "hello");
+        assert_eq!(escape_like_pattern("\\%_"), "\\\\\\%\\_");
+        assert_eq!(escape_like_pattern("%%__\\\\"), "\\%\\%\\_\\_\\\\\\\\");
+        assert_eq!(escape_like_pattern("abc123"), "abc123");
+        assert_eq!(escape_like_pattern("%_\\"), "\\%\\_\\\\");
+        assert_eq!(
+            escape_like_pattern("%%__\\\\another%string"),
+            "\\%\\%\\_\\_\\\\\\\\another\\%string"
+        );
+        assert_eq!(escape_like_pattern("foo%bar_"), "foo\\%bar\\_");
+        assert_eq!(escape_like_pattern("\\_\\%"), "\\\\\\_\\\\\\%");
+    }
+}

src/common/grpc-expr/src/alter.rs

@@ -60,6 +60,7 @@ pub fn alter_expr_to_request(table_id: TableId, expr: AlterTableExpr) -> Result<
                         column_schema: schema,
                         is_key: column_def.semantic_type == SemanticType::Tag as i32,
                         location: parse_location(ac.location)?,
+                        add_if_not_exists: ac.add_if_not_exists,
                     })
                 })
                 .collect::<Result<Vec<_>>>()?;
@@ -220,6 +221,7 @@ mod tests {
                         ..Default::default()
                     }),
                     location: None,
+                    add_if_not_exists: true,
                 }],
             })),
         };
@@ -240,6 +242,7 @@ mod tests {
             add_column.column_schema.data_type
         );
         assert_eq!(None, add_column.location);
+        assert!(add_column.add_if_not_exists);
     }
 
     #[test]
@@ -265,6 +268,7 @@ mod tests {
                             location_type: LocationType::First.into(),
                             after_column_name: String::default(),
                         }),
+                        add_if_not_exists: false,
                     },
                     AddColumn {
                         column_def: Some(ColumnDef {
@@ -280,6 +284,7 @@ mod tests {
                             location_type: LocationType::After.into(),
                             after_column_name: "ts".to_string(),
                         }),
+                        add_if_not_exists: true,
                     },
                 ],
             })),
@@ -308,6 +313,7 @@ mod tests {
             }),
             add_column.location
         );
+        assert!(add_column.add_if_not_exists);
 
         let add_column = add_columns.pop().unwrap();
         assert!(!add_column.is_key);
@@ -317,6 +323,7 @@ mod tests {
             add_column.column_schema.data_type
         );
         assert_eq!(Some(AddColumnLocation::First), add_column.location);
+        assert!(!add_column.add_if_not_exists);
     }
 
     #[test]

src/common/grpc-expr/src/insert.rs

@@ -299,6 +299,7 @@ mod tests {
                 .unwrap()
             )
         );
+        assert!(host_column.add_if_not_exists);
 
         let memory_column = &add_columns.add_columns[1];
         assert_eq!(
@@ -311,6 +312,7 @@ mod tests {
                 .unwrap()
             )
         );
+        assert!(host_column.add_if_not_exists);
 
         let time_column = &add_columns.add_columns[2];
         assert_eq!(
@@ -323,6 +325,7 @@ mod tests {
                 .unwrap()
             )
         );
+        assert!(host_column.add_if_not_exists);
 
         let interval_column = &add_columns.add_columns[3];
         assert_eq!(
@@ -335,6 +338,7 @@ mod tests {
                 .unwrap()
             )
         );
+        assert!(host_column.add_if_not_exists);
 
         let decimal_column = &add_columns.add_columns[4];
         assert_eq!(
@@ -352,6 +356,7 @@ mod tests {
                 .unwrap()
             )
         );
+        assert!(host_column.add_if_not_exists);
     }
 
     #[test]

src/common/grpc-expr/src/util.rs

@@ -192,6 +192,9 @@ pub fn build_create_table_expr(
     Ok(expr)
 }
 
+/// Find columns that are not present in the schema and return them as `AddColumns`
+/// for adding columns automatically.
+/// It always sets `add_if_not_exists` to `true` for now.
 pub fn extract_new_columns(
     schema: &Schema,
     column_exprs: Vec<ColumnExpr>,
@@ -213,6 +216,7 @@ pub fn extract_new_columns(
             AddColumn {
                 column_def,
                 location: None,
+                add_if_not_exists: true,
             }
         })
         .collect::<Vec<_>>();

src/common/meta/src/cache/container.rs

@@ -43,7 +43,7 @@ pub struct CacheContainer<K, V, CacheToken> {
     cache: Cache<K, V>,
     invalidator: Invalidator<K, V, CacheToken>,
     initializer: Initializer<K, V>,
-    token_filter: TokenFilter<CacheToken>,
+    token_filter: fn(&CacheToken) -> bool,
 }
 
 impl<K, V, CacheToken> CacheContainer<K, V, CacheToken>
@@ -58,7 +58,7 @@ where
         cache: Cache<K, V>,
         invalidator: Invalidator<K, V, CacheToken>,
         initializer: Initializer<K, V>,
-        token_filter: TokenFilter<CacheToken>,
+        token_filter: fn(&CacheToken) -> bool,
     ) -> Self {
         Self {
             name,
@@ -206,10 +206,13 @@ mod tests {
         name: &'a str,
     }
 
+    fn always_true_filter(_: &String) -> bool {
+        true
+    }
+
     #[tokio::test]
     async fn test_get() {
         let cache: Cache<NameKey, String> = CacheBuilder::new(128).build();
-        let filter: TokenFilter<String> = Box::new(|_| true);
         let counter = Arc::new(AtomicI32::new(0));
         let moved_counter = counter.clone();
         let init: Initializer<NameKey, String> = Arc::new(move |_| {
@@ -219,7 +222,13 @@ mod tests {
         let invalidator: Invalidator<NameKey, String, String> =
             Box::new(|_, _| Box::pin(async { Ok(()) }));
 
-        let adv_cache = CacheContainer::new("test".to_string(), cache, invalidator, init, filter);
+        let adv_cache = CacheContainer::new(
+            "test".to_string(),
+            cache,
+            invalidator,
+            init,
+            always_true_filter,
+        );
         let key = NameKey { name: "key" };
         let value = adv_cache.get(key).await.unwrap().unwrap();
         assert_eq!(value, "hi");
@@ -233,7 +242,6 @@ mod tests {
     #[tokio::test]
     async fn test_get_by_ref() {
         let cache: Cache<String, String> = CacheBuilder::new(128).build();
-        let filter: TokenFilter<String> = Box::new(|_| true);
         let counter = Arc::new(AtomicI32::new(0));
         let moved_counter = counter.clone();
         let init: Initializer<String, String> = Arc::new(move |_| {
@@ -243,7 +251,13 @@ mod tests {
         let invalidator: Invalidator<String, String, String> =
             Box::new(|_, _| Box::pin(async { Ok(()) }));
 
-        let adv_cache = CacheContainer::new("test".to_string(), cache, invalidator, init, filter);
+        let adv_cache = CacheContainer::new(
+            "test".to_string(),
+            cache,
+            invalidator,
+            init,
+            always_true_filter,
+        );
         let value = adv_cache.get_by_ref("foo").await.unwrap().unwrap();
         assert_eq!(value, "hi");
         let value = adv_cache.get_by_ref("foo").await.unwrap().unwrap();
@@ -257,13 +271,18 @@ mod tests {
     #[tokio::test]
     async fn test_get_value_not_exits() {
         let cache: Cache<String, String> = CacheBuilder::new(128).build();
-        let filter: TokenFilter<String> = Box::new(|_| true);
         let init: Initializer<String, String> =
             Arc::new(move |_| Box::pin(async { error::ValueNotExistSnafu {}.fail() }));
         let invalidator: Invalidator<String, String, String> =
             Box::new(|_, _| Box::pin(async { Ok(()) }));
 
-        let adv_cache = CacheContainer::new("test".to_string(), cache, invalidator, init, filter);
+        let adv_cache = CacheContainer::new(
+            "test".to_string(),
+            cache,
+            invalidator,
+            init,
+            always_true_filter,
+        );
         let value = adv_cache.get_by_ref("foo").await.unwrap();
         assert!(value.is_none());
     }
@@ -271,7 +290,6 @@ mod tests {
     #[tokio::test]
     async fn test_invalidate() {
         let cache: Cache<String, String> = CacheBuilder::new(128).build();
-        let filter: TokenFilter<String> = Box::new(|_| true);
         let counter = Arc::new(AtomicI32::new(0));
         let moved_counter = counter.clone();
         let init: Initializer<String, String> = Arc::new(move |_| {
@@ -285,7 +303,13 @@ mod tests {
             })
         });
 
-        let adv_cache = CacheContainer::new("test".to_string(), cache, invalidator, init, filter);
+        let adv_cache = CacheContainer::new(
+            "test".to_string(),
+            cache,
+            invalidator,
+            init,
+            always_true_filter,
+        );
         let value = adv_cache.get_by_ref("foo").await.unwrap().unwrap();
         assert_eq!(value, "hi");
         let value = adv_cache.get_by_ref("foo").await.unwrap().unwrap();

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

@@ -45,7 +45,7 @@ pub fn new_table_flownode_set_cache(
     let table_flow_manager = Arc::new(TableFlowManager::new(kv_backend));
     let init = init_factory(table_flow_manager);
 
-    CacheContainer::new(name, cache, Box::new(invalidator), init, Box::new(filter))
+    CacheContainer::new(name, cache, Box::new(invalidator), init, filter)
 }
 
 fn init_factory(table_flow_manager: TableFlowManagerRef) -> Initializer<TableId, FlownodeSet> {

src/common/meta/src/cache/registry.rs

@@ -151,12 +151,15 @@ mod tests {
     use crate::cache::*;
     use crate::instruction::CacheIdent;
 
+    fn always_true_filter(_: &CacheIdent) -> bool {
+        true
+    }
+
     fn test_cache(
         name: &str,
         invalidator: Invalidator<String, String, CacheIdent>,
     ) -> CacheContainer<String, String, CacheIdent> {
         let cache: Cache<String, String> = CacheBuilder::new(128).build();
-        let filter: TokenFilter<CacheIdent> = Box::new(|_| true);
         let counter = Arc::new(AtomicI32::new(0));
         let moved_counter = counter.clone();
         let init: Initializer<String, String> = Arc::new(move |_| {
@@ -164,7 +167,13 @@ mod tests {
             Box::pin(async { Ok(Some("hi".to_string())) })
         });
 
-        CacheContainer::new(name.to_string(), cache, invalidator, init, filter)
+        CacheContainer::new(
+            name.to_string(),
+            cache,
+            invalidator,
+            init,
+            always_true_filter,
+        )
     }
 
     fn test_i32_cache(
@@ -172,7 +181,6 @@ mod tests {
         invalidator: Invalidator<i32, String, CacheIdent>,
     ) -> CacheContainer<i32, String, CacheIdent> {
         let cache: Cache<i32, String> = CacheBuilder::new(128).build();
-        let filter: TokenFilter<CacheIdent> = Box::new(|_| true);
         let counter = Arc::new(AtomicI32::new(0));
         let moved_counter = counter.clone();
         let init: Initializer<i32, String> = Arc::new(move |_| {
@@ -180,7 +188,13 @@ mod tests {
             Box::pin(async { Ok(Some("foo".to_string())) })
         });
 
-        CacheContainer::new(name.to_string(), cache, invalidator, init, filter)
+        CacheContainer::new(
+            name.to_string(),
+            cache,
+            invalidator,
+            init,
+            always_true_filter,
+        )
     }
 
     #[tokio::test]

src/common/meta/src/cache/table/schema.rs

@@ -36,7 +36,7 @@ pub fn new_schema_cache(
     let schema_manager = SchemaManager::new(kv_backend.clone());
     let init = init_factory(schema_manager);
 
-    CacheContainer::new(name, cache, Box::new(invalidator), init, Box::new(filter))
+    CacheContainer::new(name, cache, Box::new(invalidator), init, filter)
 }
 
 fn init_factory(schema_manager: SchemaManager) -> Initializer<SchemaName, Arc<SchemaNameValue>> {

src/common/meta/src/cache/table/table_info.rs

@@ -41,7 +41,7 @@ pub fn new_table_info_cache(
     let table_info_manager = Arc::new(TableInfoManager::new(kv_backend));
     let init = init_factory(table_info_manager);
 
-    CacheContainer::new(name, cache, Box::new(invalidator), init, Box::new(filter))
+    CacheContainer::new(name, cache, Box::new(invalidator), init, filter)
 }
 
 fn init_factory(table_info_manager: TableInfoManagerRef) -> Initializer<TableId, Arc<TableInfo>> {

src/common/meta/src/cache/table/table_name.rs

@@ -41,7 +41,7 @@ pub fn new_table_name_cache(
     let table_name_manager = Arc::new(TableNameManager::new(kv_backend));
     let init = init_factory(table_name_manager);
 
-    CacheContainer::new(name, cache, Box::new(invalidator), init, Box::new(filter))
+    CacheContainer::new(name, cache, Box::new(invalidator), init, filter)
 }
 
 fn init_factory(table_name_manager: TableNameManagerRef) -> Initializer<TableName, TableId> {

src/common/meta/src/cache/table/table_route.rs

@@ -65,7 +65,7 @@ pub fn new_table_route_cache(
     let table_info_manager = Arc::new(TableRouteManager::new(kv_backend));
     let init = init_factory(table_info_manager);
 
-    CacheContainer::new(name, cache, Box::new(invalidator), init, Box::new(filter))
+    CacheContainer::new(name, cache, Box::new(invalidator), init, filter)
 }
 
 fn init_factory(

src/common/meta/src/cache/table/table_schema.rs

@@ -40,7 +40,7 @@ pub fn new_table_schema_cache(
     let table_info_manager = TableInfoManager::new(kv_backend);
     let init = init_factory(table_info_manager);
 
-    CacheContainer::new(name, cache, Box::new(invalidator), init, Box::new(filter))
+    CacheContainer::new(name, cache, Box::new(invalidator), init, filter)
 }
 
 fn init_factory(table_info_manager: TableInfoManager) -> Initializer<TableId, Arc<SchemaName>> {

src/common/meta/src/cache/table/view_info.rs

@@ -40,7 +40,7 @@ pub fn new_view_info_cache(
     let view_info_manager = Arc::new(ViewInfoManager::new(kv_backend));
     let init = init_factory(view_info_manager);
 
-    CacheContainer::new(name, cache, Box::new(invalidator), init, Box::new(filter))
+    CacheContainer::new(name, cache, Box::new(invalidator), init, filter)
 }
 
 fn init_factory(view_info_manager: ViewInfoManagerRef) -> Initializer<TableId, Arc<ViewInfoValue>> {

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

@@ -105,7 +105,7 @@ impl AlterLogicalTablesProcedure {
                 .context(ConvertAlterTableRequestSnafu)?;
         let new_meta = table_info
             .meta
-            .builder_with_alter_kind(table_ref.table, &request.alter_kind, true)
+            .builder_with_alter_kind(table_ref.table, &request.alter_kind)
             .context(error::TableSnafu)?
             .build()
             .with_context(|_| error::BuildTableMetaSnafu {

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

@@ -28,13 +28,13 @@ use common_procedure::error::{FromJsonSnafu, Result as ProcedureResult, ToJsonSn
 use common_procedure::{
     Context as ProcedureContext, Error as ProcedureError, LockKey, Procedure, Status, StringKey,
 };
-use common_telemetry::{debug, info};
+use common_telemetry::{debug, error, info};
 use futures::future;
 use serde::{Deserialize, Serialize};
 use snafu::ResultExt;
 use store_api::storage::RegionId;
 use strum::AsRefStr;
-use table::metadata::{RawTableInfo, TableId};
+use table::metadata::{RawTableInfo, TableId, TableInfo};
 use table::table_reference::TableReference;
 
 use crate::cache_invalidator::Context;
@@ -51,10 +51,14 @@ use crate::{metrics, ClusterId};
 
 /// The alter table procedure
 pub struct AlterTableProcedure {
-    // The runtime context.
+    /// The runtime context.
     context: DdlContext,
-    // The serialized data.
+    /// The serialized data.
     data: AlterTableData,
+    /// Cached new table metadata in the prepare step.
+    /// If we recover the procedure from json, then the table info value is not cached.
+    /// But we already validated it in the prepare step.
+    new_table_info: Option<TableInfo>,
 }
 
 impl AlterTableProcedure {
@@ -70,18 +74,31 @@ impl AlterTableProcedure {
         Ok(Self {
             context,
             data: AlterTableData::new(task, table_id, cluster_id),
+            new_table_info: None,
         })
     }
 
     pub fn from_json(json: &str, context: DdlContext) -> ProcedureResult<Self> {
         let data: AlterTableData = serde_json::from_str(json).context(FromJsonSnafu)?;
-        Ok(AlterTableProcedure { context, data })
+        Ok(AlterTableProcedure {
+            context,
+            data,
+            new_table_info: None,
+        })
     }
 
     // Checks whether the table exists.
     pub(crate) async fn on_prepare(&mut self) -> Result<Status> {
         self.check_alter().await?;
         self.fill_table_info().await?;
+
+        // Validates the request and builds the new table info.
+        // We need to build the new table info here because we should ensure the alteration
+        // is valid in `UpdateMeta` state as we already altered the region.
+        // Safety: `fill_table_info()` already set it.
+        let table_info_value = self.data.table_info_value.as_ref().unwrap();
+        self.new_table_info = Some(self.build_new_table_info(&table_info_value.table_info)?);
+
         // Safety: Checked in `AlterTableProcedure::new`.
         let alter_kind = self.data.task.alter_table.kind.as_ref().unwrap();
         if matches!(alter_kind, Kind::RenameTable { .. }) {
@@ -106,6 +123,14 @@ impl AlterTableProcedure {
 
         let leaders = find_leaders(&physical_table_route.region_routes);
         let mut alter_region_tasks = Vec::with_capacity(leaders.len());
+        let alter_kind = self.make_region_alter_kind()?;
+
+        info!(
+            "Submitting alter region requests for table {}, table_id: {}, alter_kind: {:?}",
+            self.data.table_ref(),
+            table_id,
+            alter_kind,
+        );
 
         for datanode in leaders {
             let requester = self.context.node_manager.datanode(&datanode).await;
@@ -113,7 +138,7 @@ impl AlterTableProcedure {
 
             for region in regions {
                 let region_id = RegionId::new(table_id, region);
-                let request = self.make_alter_region_request(region_id)?;
+                let request = self.make_alter_region_request(region_id, alter_kind.clone())?;
                 debug!("Submitting {request:?} to {datanode}");
 
                 let datanode = datanode.clone();
@@ -150,7 +175,15 @@ impl AlterTableProcedure {
         let table_ref = self.data.table_ref();
         // Safety: checked before.
         let table_info_value = self.data.table_info_value.as_ref().unwrap();
-        let new_info = self.build_new_table_info(&table_info_value.table_info)?;
+        // Gets the table info from the cache or builds it.
+        let new_info = match &self.new_table_info {
+            Some(cached) => cached.clone(),
+            None => self.build_new_table_info(&table_info_value.table_info)
+                .inspect_err(|e| {
+                    // We already check the table info in the prepare step so this should not happen.
+                    error!(e; "Unable to build info for table {} in update metadata step, table_id: {}", table_ref, table_id);
+                })?,
+        };
 
         debug!(
             "Starting update table: {} metadata, new table info {:?}",
@@ -174,7 +207,7 @@ impl AlterTableProcedure {
             .await?;
         }
 
-        info!("Updated table metadata for table {table_ref}, table_id: {table_id}");
+        info!("Updated table metadata for table {table_ref}, table_id: {table_id}, kind: {alter_kind:?}");
         self.data.state = AlterTableState::InvalidateTableCache;
         Ok(Status::executing(true))
     }

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

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashSet;
+
 use api::v1::alter_table_expr::Kind;
 use api::v1::region::region_request::Body;
 use api::v1::region::{
@@ -27,13 +29,15 @@ use crate::ddl::alter_table::AlterTableProcedure;
 use crate::error::{InvalidProtoMsgSnafu, Result};
 
 impl AlterTableProcedure {
-    /// Makes alter region request.
-    pub(crate) fn make_alter_region_request(&self, region_id: RegionId) -> Result<RegionRequest> {
-        // Safety: Checked in `AlterTableProcedure::new`.
-        let alter_kind = self.data.task.alter_table.kind.as_ref().unwrap();
+    /// Makes alter region request from existing an alter kind.
+    /// Region alter request always add columns if not exist.
+    pub(crate) fn make_alter_region_request(
+        &self,
+        region_id: RegionId,
+        kind: Option<alter_request::Kind>,
+    ) -> Result<RegionRequest> {
         // Safety: checked
         let table_info = self.data.table_info().unwrap();
-        let kind = create_proto_alter_kind(table_info, alter_kind)?;
 
         Ok(RegionRequest {
             header: Some(RegionRequestHeader {
@@ -47,45 +51,66 @@ impl AlterTableProcedure {
             })),
         })
     }
+
+    /// Makes alter kind proto that all regions can reuse.
+    /// Region alter request always add columns if not exist.
+    pub(crate) fn make_region_alter_kind(&self) -> Result<Option<alter_request::Kind>> {
+        // Safety: Checked in `AlterTableProcedure::new`.
+        let alter_kind = self.data.task.alter_table.kind.as_ref().unwrap();
+        // Safety: checked
+        let table_info = self.data.table_info().unwrap();
+        let kind = create_proto_alter_kind(table_info, alter_kind)?;
+
+        Ok(kind)
+    }
 }
 
 /// Creates region proto alter kind from `table_info` and `alter_kind`.
 ///
-/// Returns the kind and next column id if it adds new columns.
+/// It always adds column if not exists and drops column if exists.
+/// It skips the column if it already exists in the table.
 fn create_proto_alter_kind(
     table_info: &RawTableInfo,
     alter_kind: &Kind,
 ) -> Result<Option<alter_request::Kind>> {
     match alter_kind {
         Kind::AddColumns(x) => {
+            // Construct a set of existing columns in the table.
+            let existing_columns: HashSet<_> = table_info
+                .meta
+                .schema
+                .column_schemas
+                .iter()
+                .map(|col| &col.name)
+                .collect();
             let mut next_column_id = table_info.meta.next_column_id;
 
-            let add_columns = x
-                .add_columns
-                .iter()
-                .map(|add_column| {
-                    let column_def =
-                        add_column
-                            .column_def
-                            .as_ref()
-                            .context(InvalidProtoMsgSnafu {
-                                err_msg: "'column_def' is absent",
-                            })?;
+            let mut add_columns = Vec::with_capacity(x.add_columns.len());
+            for add_column in &x.add_columns {
+                let column_def = add_column
+                    .column_def
+                    .as_ref()
+                    .context(InvalidProtoMsgSnafu {
+                        err_msg: "'column_def' is absent",
+                    })?;
 
-                    let column_id = next_column_id;
-                    next_column_id += 1;
+                // Skips existing columns.
+                if existing_columns.contains(&column_def.name) {
+                    continue;
+                }
 
-                    let column_def = RegionColumnDef {
-                        column_def: Some(column_def.clone()),
-                        column_id,
-                    };
+                let column_id = next_column_id;
+                next_column_id += 1;
+                let column_def = RegionColumnDef {
+                    column_def: Some(column_def.clone()),
+                    column_id,
+                };
 
-                    Ok(AddColumn {
-                        column_def: Some(column_def),
-                        location: add_column.location.clone(),
-                    })
-                })
-                .collect::<Result<Vec<_>>>()?;
+                add_columns.push(AddColumn {
+                    column_def: Some(column_def),
+                    location: add_column.location.clone(),
+                });
+            }
 
             Ok(Some(alter_request::Kind::AddColumns(AddColumns {
                 add_columns,
@@ -143,6 +168,7 @@ mod tests {
     use crate::rpc::router::{Region, RegionRoute};
     use crate::test_util::{new_ddl_context, MockDatanodeManager};
 
+    /// Prepares a region with schema `[ts: Timestamp, host: Tag, cpu: Field]`.
     async fn prepare_ddl_context() -> (DdlContext, u64, TableId, RegionId, String) {
         let datanode_manager = Arc::new(MockDatanodeManager::new(()));
         let ddl_context = new_ddl_context(datanode_manager);
@@ -171,6 +197,7 @@ mod tests {
                     .name("cpu")
                     .data_type(ColumnDataType::Float64)
                     .semantic_type(SemanticType::Field)
+                    .is_nullable(true)
                     .build()
                     .unwrap()
                     .into(),
@@ -225,15 +252,16 @@ mod tests {
                             name: "my_tag3".to_string(),
                             data_type: ColumnDataType::String as i32,
                             is_nullable: true,
-                            default_constraint: b"hello".to_vec(),
+                            default_constraint: Vec::new(),
                             semantic_type: SemanticType::Tag as i32,
                             comment: String::new(),
                             ..Default::default()
                         }),
                         location: Some(AddColumnLocation {
                             location_type: LocationType::After as i32,
-                            after_column_name: "my_tag2".to_string(),
+                            after_column_name: "host".to_string(),
                         }),
+                        add_if_not_exists: false,
                     }],
                 })),
             },
@@ -242,8 +270,11 @@ mod tests {
         let mut procedure =
             AlterTableProcedure::new(cluster_id, table_id, task, ddl_context).unwrap();
         procedure.on_prepare().await.unwrap();
-        let Some(Body::Alter(alter_region_request)) =
-            procedure.make_alter_region_request(region_id).unwrap().body
+        let alter_kind = procedure.make_region_alter_kind().unwrap();
+        let Some(Body::Alter(alter_region_request)) = procedure
+            .make_alter_region_request(region_id, alter_kind)
+            .unwrap()
+            .body
         else {
             unreachable!()
         };
@@ -259,7 +290,7 @@ mod tests {
                                 name: "my_tag3".to_string(),
                                 data_type: ColumnDataType::String as i32,
                                 is_nullable: true,
-                                default_constraint: b"hello".to_vec(),
+                                default_constraint: Vec::new(),
                                 semantic_type: SemanticType::Tag as i32,
                                 comment: String::new(),
                                 ..Default::default()
@@ -268,7 +299,7 @@ mod tests {
                         }),
                         location: Some(AddColumnLocation {
                             location_type: LocationType::After as i32,
-                            after_column_name: "my_tag2".to_string(),
+                            after_column_name: "host".to_string(),
                         }),
                     }]
                 }
@@ -299,8 +330,11 @@ mod tests {
         let mut procedure =
             AlterTableProcedure::new(cluster_id, table_id, task, ddl_context).unwrap();
         procedure.on_prepare().await.unwrap();
-        let Some(Body::Alter(alter_region_request)) =
-            procedure.make_alter_region_request(region_id).unwrap().body
+        let alter_kind = procedure.make_region_alter_kind().unwrap();
+        let Some(Body::Alter(alter_region_request)) = procedure
+            .make_alter_region_request(region_id, alter_kind)
+            .unwrap()
+            .body
         else {
             unreachable!()
         };

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

@@ -23,7 +23,9 @@ use crate::key::table_info::TableInfoValue;
 use crate::key::{DeserializedValueWithBytes, RegionDistribution};
 
 impl AlterTableProcedure {
-    /// Builds new_meta
+    /// Builds new table info after alteration.
+    /// It bumps the column id of the table by the number of the add column requests.
+    /// So there may be holes in the column id sequence.
     pub(crate) fn build_new_table_info(&self, table_info: &RawTableInfo) -> Result<TableInfo> {
         let table_info =
             TableInfo::try_from(table_info.clone()).context(error::ConvertRawTableInfoSnafu)?;
@@ -34,7 +36,7 @@ impl AlterTableProcedure {
 
         let new_meta = table_info
             .meta
-            .builder_with_alter_kind(table_ref.table, &request.alter_kind, false)
+            .builder_with_alter_kind(table_ref.table, &request.alter_kind)
             .context(error::TableSnafu)?
             .build()
             .with_context(|_| error::BuildTableMetaSnafu {
@@ -46,6 +48,9 @@ impl AlterTableProcedure {
         new_info.ident.version = table_info.ident.version + 1;
         match request.alter_kind {
             AlterKind::AddColumns { columns } => {
+                // Bumps the column id for the new columns.
+                // It may bump more than the actual number of columns added if there are
+                // existing columns, but it's fine.
                 new_info.meta.next_column_id += columns.len() as u32;
             }
             AlterKind::RenameTable { new_table_name } => {

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

@@ -30,6 +30,8 @@ pub struct TestAlterTableExpr {
     add_columns: Vec<ColumnDef>,
     #[builder(setter(into, strip_option))]
     new_table_name: Option<String>,
+    #[builder(setter)]
+    add_if_not_exists: bool,
 }
 
 impl From<TestAlterTableExpr> for AlterTableExpr {
@@ -53,6 +55,7 @@ impl From<TestAlterTableExpr> for AlterTableExpr {
                         .map(|col| AddColumn {
                             column_def: Some(col),
                             location: None,
+                            add_if_not_exists: value.add_if_not_exists,
                         })
                         .collect(),
                 })),

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

@@ -56,6 +56,7 @@ fn make_alter_logical_table_add_column_task(
     let alter_table = alter_table
         .table_name(table.to_string())
         .add_columns(add_columns)
+        .add_if_not_exists(true)
         .build()
         .unwrap();
 

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

@@ -139,7 +139,7 @@ async fn test_on_submit_alter_request() {
             table_name: table_name.to_string(),
             kind: Some(Kind::DropColumns(DropColumns {
                 drop_columns: vec![DropColumn {
-                    name: "my_field_column".to_string(),
+                    name: "cpu".to_string(),
                 }],
             })),
         },
@@ -225,7 +225,7 @@ async fn test_on_submit_alter_request_with_outdated_request() {
             table_name: table_name.to_string(),
             kind: Some(Kind::DropColumns(DropColumns {
                 drop_columns: vec![DropColumn {
-                    name: "my_field_column".to_string(),
+                    name: "cpu".to_string(),
                 }],
             })),
         },
@@ -330,6 +330,7 @@ async fn test_on_update_metadata_add_columns() {
                         ..Default::default()
                     }),
                     location: None,
+                    add_if_not_exists: false,
                 }],
             })),
         },

src/common/meta/src/key.rs

@@ -137,6 +137,7 @@ use self::schema_name::{SchemaManager, SchemaNameKey, SchemaNameValue};
 use self::table_route::{TableRouteManager, TableRouteValue};
 use self::tombstone::TombstoneManager;
 use crate::error::{self, Result, SerdeJsonSnafu};
+use crate::key::flow::flow_state::FlowStateValue;
 use crate::key::node_address::NodeAddressValue;
 use crate::key::table_route::TableRouteKey;
 use crate::key::txn_helper::TxnOpGetResponseSet;
@@ -1262,7 +1263,8 @@ impl_metadata_value! {
     FlowRouteValue,
     TableFlowValue,
     NodeAddressValue,
-    SchemaNameValue
+    SchemaNameValue,
+    FlowStateValue
 }
 
 impl_optional_metadata_value! {

src/common/meta/src/key/catalog_name.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 use std::fmt::Display;
-use std::sync::Arc;
 
 use common_catalog::consts::DEFAULT_CATALOG_NAME;
 use futures::stream::BoxStream;
@@ -146,7 +145,7 @@ impl CatalogManager {
             self.kv_backend.clone(),
             req,
             DEFAULT_PAGE_SIZE,
-            Arc::new(catalog_decoder),
+            catalog_decoder,
         )
         .into_stream();
 
@@ -156,6 +155,8 @@ impl CatalogManager {
 
 #[cfg(test)]
 mod tests {
+    use std::sync::Arc;
+
     use super::*;
     use crate::kv_backend::memory::MemoryKvBackend;
 

src/common/meta/src/key/datanode_table.rs

@@ -14,7 +14,6 @@
 
 use std::collections::HashMap;
 use std::fmt::Display;
-use std::sync::Arc;
 
 use futures::stream::BoxStream;
 use serde::{Deserialize, Serialize};
@@ -166,7 +165,7 @@ impl DatanodeTableManager {
             self.kv_backend.clone(),
             req,
             DEFAULT_PAGE_SIZE,
-            Arc::new(datanode_table_value_decoder),
+            datanode_table_value_decoder,
         )
         .into_stream();
 

src/common/meta/src/key/flow.rs

@@ -15,6 +15,7 @@
 pub mod flow_info;
 pub(crate) mod flow_name;
 pub(crate) mod flow_route;
+pub mod flow_state;
 pub(crate) mod flownode_flow;
 pub(crate) mod table_flow;
 
@@ -35,6 +36,7 @@ use crate::ensure_values;
 use crate::error::{self, Result};
 use crate::key::flow::flow_info::FlowInfoManager;
 use crate::key::flow::flow_name::FlowNameManager;
+use crate::key::flow::flow_state::FlowStateManager;
 use crate::key::flow::flownode_flow::FlownodeFlowManager;
 pub use crate::key::flow::table_flow::{TableFlowManager, TableFlowManagerRef};
 use crate::key::txn_helper::TxnOpGetResponseSet;
@@ -102,6 +104,8 @@ pub struct FlowMetadataManager {
     flownode_flow_manager: FlownodeFlowManager,
     table_flow_manager: TableFlowManager,
     flow_name_manager: FlowNameManager,
+    /// only metasrv have access to itself's memory backend, so for other case it should be None
+    flow_state_manager: Option<FlowStateManager>,
     kv_backend: KvBackendRef,
 }
 
@@ -114,6 +118,7 @@ impl FlowMetadataManager {
             flow_name_manager: FlowNameManager::new(kv_backend.clone()),
             flownode_flow_manager: FlownodeFlowManager::new(kv_backend.clone()),
             table_flow_manager: TableFlowManager::new(kv_backend.clone()),
+            flow_state_manager: None,
             kv_backend,
         }
     }
@@ -123,6 +128,10 @@ impl FlowMetadataManager {
         &self.flow_name_manager
     }
 
+    pub fn flow_state_manager(&self) -> Option<&FlowStateManager> {
+        self.flow_state_manager.as_ref()
+    }
+
     /// Returns the [`FlowInfoManager`].
     pub fn flow_info_manager(&self) -> &FlowInfoManager {
         &self.flow_info_manager

src/common/meta/src/key/flow/flow_name.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
-
 use futures::stream::BoxStream;
 use lazy_static::lazy_static;
 use regex::Regex;
@@ -201,7 +199,7 @@ impl FlowNameManager {
             self.kv_backend.clone(),
             req,
             DEFAULT_PAGE_SIZE,
-            Arc::new(flow_name_decoder),
+            flow_name_decoder,
         )
         .into_stream();
 

src/common/meta/src/key/flow/flow_route.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
-
 use futures::stream::BoxStream;
 use lazy_static::lazy_static;
 use regex::Regex;
@@ -179,7 +177,7 @@ impl FlowRouteManager {
             self.kv_backend.clone(),
             req,
             DEFAULT_PAGE_SIZE,
-            Arc::new(flow_route_decoder),
+            flow_route_decoder,
         )
         .into_stream();
 

src/common/meta/src/key/flow/flow_state.rs

@@ -0,0 +1,162 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::BTreeMap;
+use std::sync::Arc;
+
+use serde::{Deserialize, Serialize};
+
+use crate::error::{self, Result};
+use crate::key::flow::FlowScoped;
+use crate::key::{FlowId, MetadataKey, MetadataValue};
+use crate::kv_backend::KvBackendRef;
+use crate::rpc::store::PutRequest;
+
+/// The entire FlowId to Flow Size's Map is stored directly in the value part of the key.
+const FLOW_STATE_KEY: &str = "state";
+
+/// The key of flow state.
+#[derive(Debug, Clone, Copy, PartialEq)]
+struct FlowStateKeyInner;
+
+impl FlowStateKeyInner {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+impl<'a> MetadataKey<'a, FlowStateKeyInner> for FlowStateKeyInner {
+    fn to_bytes(&self) -> Vec<u8> {
+        FLOW_STATE_KEY.as_bytes().to_vec()
+    }
+
+    fn from_bytes(bytes: &'a [u8]) -> Result<FlowStateKeyInner> {
+        let key = std::str::from_utf8(bytes).map_err(|e| {
+            error::InvalidMetadataSnafu {
+                err_msg: format!(
+                    "FlowInfoKeyInner '{}' is not a valid UTF8 string: {e}",
+                    String::from_utf8_lossy(bytes)
+                ),
+            }
+            .build()
+        })?;
+        if key != FLOW_STATE_KEY {
+            return Err(error::InvalidMetadataSnafu {
+                err_msg: format!("Invalid FlowStateKeyInner '{key}'"),
+            }
+            .build());
+        }
+        Ok(FlowStateKeyInner::new())
+    }
+}
+
+/// The key stores the state size of the flow.
+///
+/// The layout: `__flow/state`.
+pub struct FlowStateKey(FlowScoped<FlowStateKeyInner>);
+
+impl FlowStateKey {
+    /// Returns the [FlowStateKey].
+    pub fn new() -> FlowStateKey {
+        let inner = FlowStateKeyInner::new();
+        FlowStateKey(FlowScoped::new(inner))
+    }
+}
+
+impl Default for FlowStateKey {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl<'a> MetadataKey<'a, FlowStateKey> for FlowStateKey {
+    fn to_bytes(&self) -> Vec<u8> {
+        self.0.to_bytes()
+    }
+
+    fn from_bytes(bytes: &'a [u8]) -> Result<FlowStateKey> {
+        Ok(FlowStateKey(FlowScoped::<FlowStateKeyInner>::from_bytes(
+            bytes,
+        )?))
+    }
+}
+
+/// The value of flow state size
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct FlowStateValue {
+    /// For each key, the bytes of the state in memory
+    pub state_size: BTreeMap<FlowId, usize>,
+}
+
+impl FlowStateValue {
+    pub fn new(state_size: BTreeMap<FlowId, usize>) -> Self {
+        Self { state_size }
+    }
+}
+
+pub type FlowStateManagerRef = Arc<FlowStateManager>;
+
+/// The manager of [FlowStateKey]. Since state size changes frequently, we store it in memory.
+///
+/// This is only used in distributed mode. When meta-srv use heartbeat to update the flow stat report
+/// and frontned use get to get the latest flow stat report.
+pub struct FlowStateManager {
+    in_memory: KvBackendRef,
+}
+
+impl FlowStateManager {
+    pub fn new(in_memory: KvBackendRef) -> Self {
+        Self { in_memory }
+    }
+
+    pub async fn get(&self) -> Result<Option<FlowStateValue>> {
+        let key = FlowStateKey::new().to_bytes();
+        self.in_memory
+            .get(&key)
+            .await?
+            .map(|x| FlowStateValue::try_from_raw_value(&x.value))
+            .transpose()
+    }
+
+    pub async fn put(&self, value: FlowStateValue) -> Result<()> {
+        let key = FlowStateKey::new().to_bytes();
+        let value = value.try_as_raw_value()?;
+        let req = PutRequest::new().with_key(key).with_value(value);
+        self.in_memory.put(req).await?;
+        Ok(())
+    }
+}
+
+/// Flow's state report, send regularly through heartbeat message
+#[derive(Debug, Clone)]
+pub struct FlowStat {
+    /// For each key, the bytes of the state in memory
+    pub state_size: BTreeMap<u32, usize>,
+}
+
+impl From<FlowStateValue> for FlowStat {
+    fn from(value: FlowStateValue) -> Self {
+        Self {
+            state_size: value.state_size,
+        }
+    }
+}
+
+impl From<FlowStat> for FlowStateValue {
+    fn from(value: FlowStat) -> Self {
+        Self {
+            state_size: value.state_size,
+        }
+    }
+}

src/common/meta/src/key/flow/flownode_flow.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
-
 use futures::stream::BoxStream;
 use futures::TryStreamExt;
 use lazy_static::lazy_static;
@@ -179,7 +177,7 @@ impl FlownodeFlowManager {
             self.kv_backend.clone(),
             req,
             DEFAULT_PAGE_SIZE,
-            Arc::new(flownode_flow_key_decoder),
+            flownode_flow_key_decoder,
         )
         .into_stream();
 

src/common/meta/src/key/flow/table_flow.rs

@@ -206,7 +206,7 @@ impl TableFlowManager {
             self.kv_backend.clone(),
             req,
             DEFAULT_PAGE_SIZE,
-            Arc::new(table_flow_decoder),
+            table_flow_decoder,
         )
         .into_stream();
 

src/common/meta/src/key/schema_metadata_manager.rs

@@ -28,13 +28,10 @@ pub type SchemaMetadataManagerRef = Arc<SchemaMetadataManager>;
 pub struct SchemaMetadataManager {
     table_id_schema_cache: TableSchemaCacheRef,
     schema_cache: SchemaCacheRef,
-    #[cfg(any(test, feature = "testing"))]
-    kv_backend: crate::kv_backend::KvBackendRef,
 }
 
 impl SchemaMetadataManager {
     /// Creates a new database meta
-    #[cfg(not(any(test, feature = "testing")))]
     pub fn new(table_id_schema_cache: TableSchemaCacheRef, schema_cache: SchemaCacheRef) -> Self {
         Self {
             table_id_schema_cache,
@@ -42,20 +39,6 @@ impl SchemaMetadataManager {
         }
     }
 
-    /// Creates a new database meta
-    #[cfg(any(test, feature = "testing"))]
-    pub fn new(
-        kv_backend: crate::kv_backend::KvBackendRef,
-        table_id_schema_cache: TableSchemaCacheRef,
-        schema_cache: SchemaCacheRef,
-    ) -> Self {
-        Self {
-            table_id_schema_cache,
-            schema_cache,
-            kv_backend,
-        }
-    }
-
     /// Gets schema options by table id.
     pub async fn get_schema_options_by_table_id(
         &self,
@@ -80,6 +63,7 @@ impl SchemaMetadataManager {
         schema_name: &str,
         catalog_name: &str,
         schema_value: Option<crate::key::schema_name::SchemaNameValue>,
+        kv_backend: crate::kv_backend::KvBackendRef,
     ) {
         use table::metadata::{RawTableInfo, TableType};
         let value = crate::key::table_info::TableInfoValue::new(RawTableInfo {
@@ -91,19 +75,18 @@ impl SchemaMetadataManager {
             meta: Default::default(),
             table_type: TableType::Base,
         });
-        let table_info_manager =
-            crate::key::table_info::TableInfoManager::new(self.kv_backend.clone());
+        let table_info_manager = crate::key::table_info::TableInfoManager::new(kv_backend.clone());
         let (txn, _) = table_info_manager
             .build_create_txn(table_id, &value)
             .unwrap();
-        let resp = self.kv_backend.txn(txn).await.unwrap();
+        let resp = kv_backend.txn(txn).await.unwrap();
         assert!(resp.succeeded, "Failed to create table metadata");
         let key = crate::key::schema_name::SchemaNameKey {
             catalog: catalog_name,
             schema: schema_name,
         };
 
-        crate::key::schema_name::SchemaManager::new(self.kv_backend.clone())
+        crate::key::schema_name::SchemaManager::new(kv_backend.clone())
             .create(key, schema_value, false)
             .await
             .expect("Failed to create schema metadata");

src/common/meta/src/key/schema_name.rs

@@ -14,7 +14,6 @@
 
 use std::collections::HashMap;
 use std::fmt::Display;
-use std::sync::Arc;
 
 use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_time::DatabaseTimeToLive;
@@ -283,7 +282,7 @@ impl SchemaManager {
             self.kv_backend.clone(),
             req,
             DEFAULT_PAGE_SIZE,
-            Arc::new(schema_decoder),
+            schema_decoder,
         )
         .into_stream();
 
@@ -308,6 +307,7 @@ impl<'a> From<&'a SchemaName> for SchemaNameKey<'a> {
 
 #[cfg(test)]
 mod tests {
+    use std::sync::Arc;
     use std::time::Duration;
 
     use super::*;

src/common/meta/src/key/table_name.rs

@@ -269,7 +269,7 @@ impl TableNameManager {
             self.kv_backend.clone(),
             req,
             DEFAULT_PAGE_SIZE,
-            Arc::new(table_decoder),
+            table_decoder,
         )
         .into_stream();
 

src/common/meta/src/kv_backend.rs

@@ -36,7 +36,7 @@ pub mod postgres;
 pub mod test;
 pub mod txn;
 
-pub type KvBackendRef = Arc<dyn KvBackend<Error = Error> + Send + Sync>;
+pub type KvBackendRef<E = Error> = Arc<dyn KvBackend<Error = E> + Send + Sync>;
 
 #[async_trait]
 pub trait KvBackend: TxnService
@@ -161,6 +161,9 @@ where
     Self::Error: ErrorExt,
 {
     fn reset(&self);
+
+    /// Upcast as `KvBackendRef`. Since https://github.com/rust-lang/rust/issues/65991 is not yet stable.
+    fn as_kv_backend_ref(self: Arc<Self>) -> KvBackendRef<Self::Error>;
 }
 
-pub type ResettableKvBackendRef = Arc<dyn ResettableKvBackend<Error = Error> + Send + Sync>;
+pub type ResettableKvBackendRef<E = Error> = Arc<dyn ResettableKvBackend<Error = E> + Send + Sync>;

src/common/meta/src/kv_backend/memory.rs

@@ -16,13 +16,13 @@ use std::any::Any;
 use std::collections::BTreeMap;
 use std::fmt::{Display, Formatter};
 use std::marker::PhantomData;
-use std::sync::RwLock;
+use std::sync::{Arc, RwLock};
 
 use async_trait::async_trait;
 use common_error::ext::ErrorExt;
 use serde::Serializer;
 
-use super::ResettableKvBackend;
+use super::{KvBackendRef, ResettableKvBackend};
 use crate::kv_backend::txn::{Txn, TxnOp, TxnOpResponse, TxnRequest, TxnResponse};
 use crate::kv_backend::{KvBackend, TxnService};
 use crate::metrics::METRIC_META_TXN_REQUEST;
@@ -311,6 +311,10 @@ impl<T: ErrorExt + Send + Sync + 'static> ResettableKvBackend for MemoryKvBacken
     fn reset(&self) {
         self.clear();
     }
+
+    fn as_kv_backend_ref(self: Arc<Self>) -> KvBackendRef<T> {
+        self
+    }
 }
 
 #[cfg(test)]

src/common/meta/src/kv_backend/postgres.rs

@@ -16,6 +16,7 @@ use std::any::Any;
 use std::borrow::Cow;
 use std::sync::Arc;
 
+use common_telemetry::error;
 use snafu::ResultExt;
 use tokio_postgres::types::ToSql;
 use tokio_postgres::{Client, NoTls};
@@ -97,7 +98,11 @@ impl PgStore {
         let (client, conn) = tokio_postgres::connect(url, NoTls)
             .await
             .context(ConnectPostgresSnafu)?;
-        tokio::spawn(async move { conn.await.context(ConnectPostgresSnafu) });
+        tokio::spawn(async move {
+            if let Err(e) = conn.await {
+                error!(e; "connection error");
+            }
+        });
         Self::with_pg_client(client).await
     }
 

src/common/meta/src/range_stream.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
-
 use async_stream::try_stream;
 use common_telemetry::debug;
 use futures::Stream;
@@ -148,7 +146,7 @@ impl PaginationStreamFactory {
 }
 
 pub struct PaginationStream<T> {
-    decoder_fn: Arc<KeyValueDecoderFn<T>>,
+    decoder_fn: fn(KeyValue) -> Result<T>,
     factory: PaginationStreamFactory,
 }
 
@@ -158,7 +156,7 @@ impl<T> PaginationStream<T> {
         kv: KvBackendRef,
         req: RangeRequest,
         page_size: usize,
-        decoder_fn: Arc<KeyValueDecoderFn<T>>,
+        decoder_fn: fn(KeyValue) -> Result<T>,
     ) -> Self {
         Self {
             decoder_fn,
@@ -191,6 +189,7 @@ mod tests {
 
     use std::assert_matches::assert_matches;
     use std::collections::BTreeMap;
+    use std::sync::Arc;
 
     use futures::TryStreamExt;
 
@@ -250,7 +249,7 @@ mod tests {
                 ..Default::default()
             },
             DEFAULT_PAGE_SIZE,
-            Arc::new(decoder),
+            decoder,
         )
         .into_stream();
         let kv = stream.try_collect::<Vec<_>>().await.unwrap();
@@ -290,7 +289,7 @@ mod tests {
                 ..Default::default()
             },
             2,
-            Arc::new(decoder),
+            decoder,
         );
         let kv = stream
             .into_stream()

src/common/meta/src/state_store.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
-
 use async_trait::async_trait;
 use common_error::ext::BoxedError;
 use common_procedure::error::{DeleteStatesSnafu, ListStateSnafu, PutStateSnafu};
@@ -171,7 +169,7 @@ impl StateStore for KvStateStore {
             self.kv_backend.clone(),
             req,
             self.max_num_per_range_request.unwrap_or_default(),
-            Arc::new(decode_kv),
+            decode_kv,
         )
         .into_stream();
 

src/common/runtime/Cargo.toml

@@ -39,3 +39,7 @@ tokio-util.workspace = true
 
 [dev-dependencies]
 tokio-test = "0.4"
+
+[target.'cfg(tokio_unstable)'.dependencies]
+tokio-metrics = { version = "0.3" }
+tokio-metrics-collector = { version = "0.2" }

src/datanode/src/datanode.rs

@@ -224,7 +224,6 @@ impl DatanodeBuilder {
             cache_registry.get().context(MissingCacheSnafu)?;
 
         let schema_metadata_manager = Arc::new(SchemaMetadataManager::new(
-            kv_backend.clone(),
             table_id_schema_cache,
             schema_cache,
         ));

src/datanode/src/store.rs

@@ -28,7 +28,7 @@ use common_telemetry::{info, warn};
 use object_store::layers::{LruCacheLayer, RetryInterceptor, RetryLayer};
 use object_store::services::Fs;
 use object_store::util::{join_dir, normalize_dir, with_instrument_layers};
-use object_store::{Access, Error, HttpClient, ObjectStore, ObjectStoreBuilder, OBJECT_CACHE_DIR};
+use object_store::{Access, Error, HttpClient, ObjectStore, ObjectStoreBuilder};
 use snafu::prelude::*;
 
 use crate::config::{HttpClientConfig, ObjectStoreConfig, DEFAULT_OBJECT_STORE_CACHE_SIZE};
@@ -147,12 +147,10 @@ async fn build_cache_layer(
     };
 
     // Enable object cache by default
-    // Set the cache_path to be `${data_home}/object_cache/read/{name}` by default
+    // Set the cache_path to be `${data_home}` by default
     // if it's not present
     if cache_path.is_none() {
-        let object_cache_path = join_dir(data_home, OBJECT_CACHE_DIR);
-        let read_cache_path = join_dir(&object_cache_path, "read");
-        let read_cache_path = join_dir(&read_cache_path, &name.to_lowercase());
+        let read_cache_path = data_home.to_string();
         tokio::fs::create_dir_all(Path::new(&read_cache_path))
             .await
             .context(CreateDirSnafu {

src/datatypes/src/schema.rs

@@ -29,7 +29,7 @@ use crate::error::{self, DuplicateColumnSnafu, Error, ProjectArrowSchemaSnafu, R
 use crate::prelude::ConcreteDataType;
 pub use crate::schema::column_schema::{
     ColumnSchema, FulltextAnalyzer, FulltextOptions, Metadata, SkippingIndexOptions,
-    COLUMN_FULLTEXT_CHANGE_OPT_KEY_ENABLE, COLUMN_FULLTEXT_OPT_KEY_ANALYZER,
+    SkippingIndexType, COLUMN_FULLTEXT_CHANGE_OPT_KEY_ENABLE, COLUMN_FULLTEXT_OPT_KEY_ANALYZER,
     COLUMN_FULLTEXT_OPT_KEY_CASE_SENSITIVE, COLUMN_SKIPPING_INDEX_OPT_KEY_GRANULARITY,
     COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE, COMMENT_KEY, FULLTEXT_KEY, INVERTED_INDEX_KEY,
     SKIPPING_INDEX_KEY, TIME_INDEX_KEY,

src/datatypes/src/schema/column_schema.rs

@@ -123,6 +123,14 @@ impl ColumnSchema {
         self.default_constraint.as_ref()
     }
 
+    /// Check if the default constraint is a impure function.
+    pub fn is_default_impure(&self) -> bool {
+        self.default_constraint
+            .as_ref()
+            .map(|c| c.is_function())
+            .unwrap_or(false)
+    }
+
     #[inline]
     pub fn metadata(&self) -> &Metadata {
         &self.metadata
@@ -283,6 +291,15 @@ impl ColumnSchema {
         }
     }
 
+    /// Creates an impure default value for this column, only if it have a impure default constraint.
+    /// Otherwise, returns `Ok(None)`.
+    pub fn create_impure_default(&self) -> Result<Option<Value>> {
+        match &self.default_constraint {
+            Some(c) => c.create_impure_default(&self.data_type),
+            None => Ok(None),
+        }
+    }
+
     /// Retrieves the fulltext options for the column.
     pub fn fulltext_options(&self) -> Result<Option<FulltextOptions>> {
         match self.metadata.get(FULLTEXT_KEY) {
@@ -543,7 +560,7 @@ pub struct SkippingIndexOptions {
     pub granularity: u32,
     /// The type of the skip index.
     #[serde(default)]
-    pub index_type: SkipIndexType,
+    pub index_type: SkippingIndexType,
 }
 
 impl fmt::Display for SkippingIndexOptions {
@@ -556,15 +573,15 @@ impl fmt::Display for SkippingIndexOptions {
 
 /// Skip index types.
 #[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize, Visit, VisitMut)]
-pub enum SkipIndexType {
+pub enum SkippingIndexType {
     #[default]
     BloomFilter,
 }
 
-impl fmt::Display for SkipIndexType {
+impl fmt::Display for SkippingIndexType {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            SkipIndexType::BloomFilter => write!(f, "BLOOM"),
+            SkippingIndexType::BloomFilter => write!(f, "BLOOM"),
         }
     }
 }
@@ -587,7 +604,7 @@ impl TryFrom<HashMap<String, String>> for SkippingIndexOptions {
         // Parse index type with default value BloomFilter
         let index_type = match options.get(COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE) {
             Some(typ) => match typ.to_ascii_uppercase().as_str() {
-                "BLOOM" => SkipIndexType::BloomFilter,
+                "BLOOM" => SkippingIndexType::BloomFilter,
                 _ => {
                     return error::InvalidSkippingIndexOptionSnafu {
                         msg: format!("Invalid index type: {typ}, expected: 'BLOOM'"),
@@ -595,7 +612,7 @@ impl TryFrom<HashMap<String, String>> for SkippingIndexOptions {
                     .fail();
                 }
             },
-            None => SkipIndexType::default(),
+            None => SkippingIndexType::default(),
         };
 
         Ok(SkippingIndexOptions {

src/datatypes/src/schema/constraint.rs

@@ -178,12 +178,63 @@ impl ColumnDefaultConstraint {
         }
     }
 
+    /// Only create default vector if it's impure, i.e., it's a function.
+    ///
+    /// This helps to delay creating constant default values to mito engine while also keeps impure default have consistent values
+    pub fn create_impure_default_vector(
+        &self,
+        data_type: &ConcreteDataType,
+        num_rows: usize,
+    ) -> Result<Option<VectorRef>> {
+        assert!(num_rows > 0);
+
+        match self {
+            ColumnDefaultConstraint::Function(expr) => {
+                // Functions should also ensure its return value is not null when
+                // is_nullable is true.
+                match &expr[..] {
+                    // TODO(dennis): we only supports current_timestamp right now,
+                    //   it's better to use a expression framework in future.
+                    CURRENT_TIMESTAMP | CURRENT_TIMESTAMP_FN | NOW_FN => {
+                        create_current_timestamp_vector(data_type, num_rows).map(Some)
+                    }
+                    _ => error::UnsupportedDefaultExprSnafu { expr }.fail(),
+                }
+            }
+            ColumnDefaultConstraint::Value(_) => Ok(None),
+        }
+    }
+
+    /// Only create default value if it's impure, i.e., it's a function.
+    ///
+    /// This helps to delay creating constant default values to mito engine while also keeps impure default have consistent values
+    pub fn create_impure_default(&self, data_type: &ConcreteDataType) -> Result<Option<Value>> {
+        match self {
+            ColumnDefaultConstraint::Function(expr) => {
+                // Functions should also ensure its return value is not null when
+                // is_nullable is true.
+                match &expr[..] {
+                    CURRENT_TIMESTAMP | CURRENT_TIMESTAMP_FN | NOW_FN => {
+                        create_current_timestamp(data_type).map(Some)
+                    }
+                    _ => error::UnsupportedDefaultExprSnafu { expr }.fail(),
+                }
+            }
+            ColumnDefaultConstraint::Value(_) => Ok(None),
+        }
+    }
+
     /// Returns true if this constraint might creates NULL.
     fn maybe_null(&self) -> bool {
         // Once we support more functions, we may return true if given function
         // could return null.
         matches!(self, ColumnDefaultConstraint::Value(Value::Null))
     }
+
+    /// Returns true if this constraint is a function.
+    pub fn is_function(&self) -> bool {
+        matches!(self, ColumnDefaultConstraint::Function(_))
+    }
 }
 
 fn create_current_timestamp(data_type: &ConcreteDataType) -> Result<Value> {

src/flow/Cargo.toml

@@ -40,9 +40,12 @@ datatypes.workspace = true
 enum-as-inner = "0.6.0"
 enum_dispatch = "0.3"
 futures = "0.3"
+get-size-derive2 = "0.1.2"
+get-size2 = "0.1.2"
 greptime-proto.workspace = true
 # This fork of hydroflow is simply for keeping our dependency in our org, and pin the version
 # otherwise it is the same with upstream repo
+http.workspace = true
 hydroflow = { git = "https://github.com/GreptimeTeam/hydroflow.git", branch = "main" }
 itertools.workspace = true
 lazy_static.workspace = true

src/flow/src/adapter.rs

@@ -30,7 +30,7 @@ use common_telemetry::{debug, info, trace};
 use datatypes::schema::ColumnSchema;
 use datatypes::value::Value;
 use greptime_proto::v1;
-use itertools::Itertools;
+use itertools::{EitherOrBoth, Itertools};
 use meta_client::MetaClientOptions;
 use query::QueryEngine;
 use serde::{Deserialize, Serialize};
@@ -45,30 +45,29 @@ use tokio::sync::broadcast::error::TryRecvError;
 use tokio::sync::{broadcast, watch, Mutex, RwLock};
 
 pub(crate) use crate::adapter::node_context::FlownodeContext;
-use crate::adapter::table_source::TableSource;
-use crate::adapter::util::column_schemas_to_proto;
+use crate::adapter::table_source::ManagedTableSource;
+use crate::adapter::util::relation_desc_to_column_schemas_with_fallback;
 use crate::adapter::worker::{create_worker, Worker, WorkerHandle};
 use crate::compute::ErrCollector;
 use crate::df_optimizer::sql_to_flow_plan;
-use crate::error::{
-    EvalSnafu, ExternalSnafu, FlowAlreadyExistSnafu, InternalSnafu, TableNotFoundSnafu,
-    UnexpectedSnafu,
-};
-use crate::expr::{Batch, GlobalId};
-use crate::metrics::{METRIC_FLOW_INSERT_ELAPSED, METRIC_FLOW_RUN_INTERVAL_MS};
-use crate::repr::{self, DiffRow, Row, BATCH_SIZE};
+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::repr::{self, DiffRow, RelationDesc, Row, BATCH_SIZE};
 
 mod flownode_impl;
 mod parse_expr;
+mod stat;
 #[cfg(test)]
 mod tests;
 mod util;
 mod worker;
 
 pub(crate) mod node_context;
-mod table_source;
+pub(crate) mod table_source;
 
 use crate::error::Error;
+use crate::utils::StateReportHandler;
 use crate::FrontendInvoker;
 
 // `GREPTIME_TIMESTAMP` is not used to distinguish when table is created automatically by flow
@@ -125,7 +124,7 @@ pub struct FlowWorkerManager {
     /// The query engine that will be used to parse the query and convert it to a dataflow plan
     pub query_engine: Arc<dyn QueryEngine>,
     /// Getting table name and table schema from table info manager
-    table_info_source: TableSource,
+    table_info_source: ManagedTableSource,
     frontend_invoker: RwLock<Option<FrontendInvoker>>,
     /// contains mapping from table name to global id, and table schema
     node_context: RwLock<FlownodeContext>,
@@ -137,6 +136,8 @@ pub struct FlowWorkerManager {
     ///
     /// So that a series of event like `inserts -> flush` can be handled correctly
     flush_lock: RwLock<()>,
+    /// receive a oneshot sender to send state size report
+    state_report_handler: RwLock<Option<StateReportHandler>>,
 }
 
 /// Building FlownodeManager
@@ -152,11 +153,11 @@ impl FlowWorkerManager {
         query_engine: Arc<dyn QueryEngine>,
         table_meta: TableMetadataManagerRef,
     ) -> Self {
-        let srv_map = TableSource::new(
+        let srv_map = ManagedTableSource::new(
             table_meta.table_info_manager().clone(),
             table_meta.table_name_manager().clone(),
         );
-        let node_context = FlownodeContext::default();
+        let node_context = FlownodeContext::new(Box::new(srv_map.clone()) as _);
         let tick_manager = FlowTickManager::new();
         let worker_handles = Vec::new();
         FlowWorkerManager {
@@ -170,9 +171,15 @@ impl FlowWorkerManager {
             tick_manager,
             node_id,
             flush_lock: RwLock::new(()),
+            state_report_handler: RwLock::new(None),
         }
     }
 
+    pub async fn with_state_report_handler(self, handler: StateReportHandler) -> Self {
+        *self.state_report_handler.write().await = Some(handler);
+        self
+    }
+
     /// Create a flownode manager with one worker
     pub fn new_with_worker<'s>(
         node_id: Option<u32>,
@@ -235,16 +242,26 @@ impl FlowWorkerManager {
             let (catalog, schema) = (table_name[0].clone(), table_name[1].clone());
             let ctx = Arc::new(QueryContext::with(&catalog, &schema));
 
-            let (is_ts_placeholder, proto_schema) =
-                self.try_fetch_or_create_table(&table_name).await?;
+            let (is_ts_placeholder, proto_schema) = self
+                .try_fetch_existing_table(&table_name)
+                .await?
+                .context(UnexpectedSnafu {
+                    reason: format!("Table not found: {}", table_name.join(".")),
+                })?;
             let schema_len = proto_schema.len();
 
+            let total_rows = reqs.iter().map(|r| r.len()).sum::<usize>();
             trace!(
                 "Sending {} writeback requests to table {}, reqs total rows={}",
                 reqs.len(),
                 table_name.join("."),
                 reqs.iter().map(|r| r.len()).sum::<usize>()
             );
+
+            METRIC_FLOW_ROWS
+                .with_label_values(&["out"])
+                .inc_by(total_rows as u64);
+
             let now = self.tick_manager.tick();
             for req in reqs {
                 match req {
@@ -380,16 +397,14 @@ impl FlowWorkerManager {
         Ok(output)
     }
 
-    /// Fetch table info or create table from flow's schema if not exist
-    async fn try_fetch_or_create_table(
+    /// Fetch table schema and primary key from table info source, if table not exist return None
+    async fn fetch_table_pk_schema(
         &self,
         table_name: &TableName,
-    ) -> Result<(bool, Vec<api::v1::ColumnSchema>), Error> {
-        // TODO(discord9): instead of auto build table from request schema, actually build table
-        // before `create flow` to be able to assign pk and ts etc.
-        let (primary_keys, schema, is_ts_placeholder) = if let Some(table_id) = self
+    ) -> Result<Option<(Vec<String>, Option<usize>, Vec<ColumnSchema>)>, Error> {
+        if let Some(table_id) = self
             .table_info_source
-            .get_table_id_from_name(table_name)
+            .get_opt_table_id_from_name(table_name)
             .await?
         {
             let table_info = self
@@ -404,102 +419,90 @@ impl FlowWorkerManager {
                 .map(|i| meta.schema.column_schemas[i].name.clone())
                 .collect_vec();
             let schema = meta.schema.column_schemas;
-            // check if the last column is the auto created timestamp column, hence the table is auto created from
-            // flow's plan type
-            let is_auto_create = {
-                let correct_name = schema
-                    .last()
-                    .map(|s| s.name == AUTO_CREATED_PLACEHOLDER_TS_COL)
-                    .unwrap_or(false);
-                let correct_time_index = meta.schema.timestamp_index == Some(schema.len() - 1);
-                correct_name && correct_time_index
-            };
-            (primary_keys, schema, is_auto_create)
+            let time_index = meta.schema.timestamp_index;
+            Ok(Some((primary_keys, time_index, schema)))
         } else {
-            // TODO(discord9): condiser remove buggy auto create by schema
+            Ok(None)
+        }
+    }
 
-            let node_ctx = self.node_context.read().await;
-            let gid: GlobalId = node_ctx
-                .table_repr
-                .get_by_name(table_name)
-                .map(|x| x.1)
-                .unwrap();
-            let schema = node_ctx
-                .schema
-                .get(&gid)
-                .with_context(|| TableNotFoundSnafu {
-                    name: format!("Table name = {:?}", table_name),
-                })?
-                .clone();
-            // TODO(discord9): use default key from schema
-            let primary_keys = schema
-                .typ()
-                .keys
-                .first()
-                .map(|v| {
-                    v.column_indices
-                        .iter()
-                        .map(|i| {
-                            schema
-                                .get_name(*i)
-                                .clone()
-                                .unwrap_or_else(|| format!("col_{i}"))
-                        })
-                        .collect_vec()
-                })
-                .unwrap_or_default();
-            let update_at = ColumnSchema::new(
-                UPDATE_AT_TS_COL,
+    /// return (primary keys, schema and if the table have a placeholder timestamp column)
+    /// schema of the table comes from flow's output plan
+    ///
+    /// adjust to add `update_at` column and ts placeholder if needed
+    async fn adjust_auto_created_table_schema(
+        &self,
+        schema: &RelationDesc,
+    ) -> Result<(Vec<String>, Vec<ColumnSchema>, bool), Error> {
+        // TODO(discord9): condiser remove buggy auto create by schema
+
+        // TODO(discord9): use default key from schema
+        let primary_keys = schema
+            .typ()
+            .keys
+            .first()
+            .map(|v| {
+                v.column_indices
+                    .iter()
+                    .map(|i| {
+                        schema
+                            .get_name(*i)
+                            .clone()
+                            .unwrap_or_else(|| format!("col_{i}"))
+                    })
+                    .collect_vec()
+            })
+            .unwrap_or_default();
+        let update_at = ColumnSchema::new(
+            UPDATE_AT_TS_COL,
+            ConcreteDataType::timestamp_millisecond_datatype(),
+            true,
+        );
+
+        let original_schema = relation_desc_to_column_schemas_with_fallback(schema);
+
+        let mut with_auto_added_col = original_schema.clone();
+        with_auto_added_col.push(update_at);
+
+        // if no time index, add one as placeholder
+        let no_time_index = schema.typ().time_index.is_none();
+        if no_time_index {
+            let ts_col = ColumnSchema::new(
+                AUTO_CREATED_PLACEHOLDER_TS_COL,
                 ConcreteDataType::timestamp_millisecond_datatype(),
                 true,
-            );
+            )
+            .with_time_index(true);
+            with_auto_added_col.push(ts_col);
+        }
 
-            let original_schema = schema
-                .typ()
-                .column_types
-                .clone()
-                .into_iter()
-                .enumerate()
-                .map(|(idx, typ)| {
-                    let name = schema
-                        .names
-                        .get(idx)
-                        .cloned()
-                        .flatten()
-                        .unwrap_or(format!("col_{}", idx));
-                    let ret = ColumnSchema::new(name, typ.scalar_type, typ.nullable);
-                    if schema.typ().time_index == Some(idx) {
-                        ret.with_time_index(true)
-                    } else {
-                        ret
-                    }
-                })
-                .collect_vec();
-
-            let mut with_auto_added_col = original_schema.clone();
-            with_auto_added_col.push(update_at);
-
-            // if no time index, add one as placeholder
-            let no_time_index = schema.typ().time_index.is_none();
-            if no_time_index {
-                let ts_col = ColumnSchema::new(
-                    AUTO_CREATED_PLACEHOLDER_TS_COL,
-                    ConcreteDataType::timestamp_millisecond_datatype(),
-                    true,
-                )
-                .with_time_index(true);
-                with_auto_added_col.push(ts_col);
-            }
-
-            (primary_keys, with_auto_added_col, no_time_index)
-        };
-        let proto_schema = column_schemas_to_proto(schema, &primary_keys)?;
-        Ok((is_ts_placeholder, proto_schema))
+        Ok((primary_keys, with_auto_added_col, no_time_index))
     }
 }
 
 /// Flow Runtime related methods
 impl FlowWorkerManager {
+    /// Start state report handler, which will receive a sender from HeartbeatTask to send state size report back
+    ///
+    /// if heartbeat task is shutdown, this future will exit too
+    async fn start_state_report_handler(self: Arc<Self>) -> Option<JoinHandle<()>> {
+        let state_report_handler = self.state_report_handler.write().await.take();
+        if let Some(mut handler) = state_report_handler {
+            let zelf = self.clone();
+            let handler = common_runtime::spawn_global(async move {
+                while let Some(ret_handler) = handler.recv().await {
+                    let state_report = zelf.gen_state_report().await;
+                    ret_handler.send(state_report).unwrap_or_else(|err| {
+                        common_telemetry::error!(err; "Send state size report error");
+                    });
+                }
+            });
+            Some(handler)
+        } else {
+            None
+        }
+    }
+
     /// run in common_runtime background runtime
     pub fn run_background(
         self: Arc<Self>,
@@ -507,6 +510,7 @@ impl FlowWorkerManager {
     ) -> JoinHandle<()> {
         info!("Starting flownode manager's background task");
         common_runtime::spawn_global(async move {
+            let _state_report_handler = self.clone().start_state_report_handler().await;
             self.run(shutdown).await;
         })
     }
@@ -533,6 +537,8 @@ impl FlowWorkerManager {
         let default_interval = Duration::from_secs(1);
         let mut avg_spd = 0; // rows/sec
         let mut since_last_run = tokio::time::Instant::now();
+        let run_per_trace = 10;
+        let mut run_cnt = 0;
         loop {
             // TODO(discord9): only run when new inputs arrive or scheduled to
             let row_cnt = self.run_available(true).await.unwrap_or_else(|err| {
@@ -575,10 +581,19 @@ impl FlowWorkerManager {
             } else {
                 (9 * avg_spd + cur_spd) / 10
             };
-            trace!("avg_spd={} r/s, cur_spd={} r/s", avg_spd, cur_spd);
             let new_wait = BATCH_SIZE * 1000 / avg_spd.max(1); //in ms
             let new_wait = Duration::from_millis(new_wait as u64).min(default_interval);
-            trace!("Wait for {} ms, row_cnt={}", new_wait.as_millis(), row_cnt);
+
+            // print trace every `run_per_trace` times so that we can see if there is something wrong
+            // but also not get flooded with trace
+            if run_cnt >= run_per_trace {
+                trace!("avg_spd={} r/s, cur_spd={} r/s", avg_spd, cur_spd);
+                trace!("Wait for {} ms, row_cnt={}", new_wait.as_millis(), row_cnt);
+                run_cnt = 0;
+            } else {
+                run_cnt += 1;
+            }
+
             METRIC_FLOW_RUN_INTERVAL_MS.set(new_wait.as_millis() as i64);
             since_last_run = tokio::time::Instant::now();
             tokio::time::sleep(new_wait).await;
@@ -638,13 +653,18 @@ impl FlowWorkerManager {
         &self,
         region_id: RegionId,
         rows: Vec<DiffRow>,
+        batch_datatypes: &[ConcreteDataType],
     ) -> Result<(), Error> {
         let rows_len = rows.len();
         let table_id = region_id.table_id();
         let _timer = METRIC_FLOW_INSERT_ELAPSED
             .with_label_values(&[table_id.to_string().as_str()])
             .start_timer();
-        self.node_context.read().await.send(table_id, rows).await?;
+        self.node_context
+            .read()
+            .await
+            .send(table_id, rows, batch_datatypes)
+            .await?;
         trace!(
             "Handling write request for table_id={} with {} rows",
             table_id,
@@ -704,43 +724,6 @@ impl FlowWorkerManager {
             query_ctx,
         } = args;
 
-        let already_exist = {
-            let mut flag = false;
-
-            // check if the task already exists
-            for handle in self.worker_handles.iter() {
-                if handle.lock().await.contains_flow(flow_id).await? {
-                    flag = true;
-                    break;
-                }
-            }
-            flag
-        };
-        match (create_if_not_exists, or_replace, already_exist) {
-            // do replace
-            (_, true, true) => {
-                info!("Replacing flow with id={}", flow_id);
-                self.remove_flow(flow_id).await?;
-            }
-            (false, false, true) => FlowAlreadyExistSnafu { id: flow_id }.fail()?,
-            // do nothing if exists
-            (true, false, true) => {
-                info!("Flow with id={} already exists, do nothing", flow_id);
-                return Ok(None);
-            }
-            // create if not exists
-            (_, _, false) => (),
-        }
-
-        if create_if_not_exists {
-            // check if the task already exists
-            for handle in self.worker_handles.iter() {
-                if handle.lock().await.contains_flow(flow_id).await? {
-                    return Ok(None);
-                }
-            }
-        }
-
         let mut node_ctx = self.node_context.write().await;
         // assign global id to source and sink table
         for source in &source_table_ids {
@@ -759,7 +742,67 @@ impl FlowWorkerManager {
         let flow_plan = sql_to_flow_plan(&mut node_ctx, &self.query_engine, &sql).await?;
 
         debug!("Flow {:?}'s Plan is {:?}", flow_id, flow_plan);
-        node_ctx.assign_table_schema(&sink_table_name, flow_plan.schema.clone())?;
+
+        // check schema against actual table schema if exists
+        // if not exist create sink table immediately
+        if let Some((_, _, real_schema)) = self.fetch_table_pk_schema(&sink_table_name).await? {
+            let auto_schema = relation_desc_to_column_schemas_with_fallback(&flow_plan.schema);
+
+            // for column schema, only `data_type` need to be check for equality
+            // since one can omit flow's column name when write flow query
+            // print a user friendly error message about mismatch and how to correct them
+            for (idx, zipped) in auto_schema
+                .iter()
+                .zip_longest(real_schema.iter())
+                .enumerate()
+            {
+                match zipped {
+                    EitherOrBoth::Both(auto, real) => {
+                        if auto.data_type != real.data_type {
+                            InvalidQuerySnafu {
+                                    reason: format!(
+                                        "Column {}(name is '{}', flow inferred name is '{}')'s data type mismatch, expect {:?} got {:?}",
+                                        idx,
+                                        real.name,
+                                        auto.name,
+                                        real.data_type,
+                                        auto.data_type
+                                    ),
+                                }
+                                .fail()?;
+                        }
+                    }
+                    EitherOrBoth::Right(real) if real.data_type.is_timestamp() => {
+                        // if table is auto created, the last one or two column should be timestamp(update at and ts placeholder)
+                        continue;
+                    }
+                    _ => InvalidQuerySnafu {
+                        reason: format!(
+                            "schema length mismatched, expected {} found {}",
+                            real_schema.len(),
+                            auto_schema.len()
+                        ),
+                    }
+                    .fail()?,
+                }
+            }
+        } else {
+            // assign inferred schema to sink table
+            // create sink table
+            let did_create = self
+                .create_table_from_relation(
+                    &format!("flow-id={flow_id}"),
+                    &sink_table_name,
+                    &flow_plan.schema,
+                )
+                .await?;
+            if !did_create {
+                UnexpectedSnafu {
+                    reason: format!("Failed to create table {:?}", sink_table_name),
+                }
+                .fail()?;
+            }
+        }
 
         let _ = comment;
         let _ = flow_options;
@@ -794,9 +837,11 @@ impl FlowWorkerManager {
             source_ids,
             src_recvs: source_receivers,
             expire_after,
+            or_replace,
             create_if_not_exists,
             err_collector,
         };
+
         handle.create_flow(create_request).await?;
         info!("Successfully create flow with id={}", flow_id);
         Ok(Some(flow_id))

src/flow/src/adapter/flownode_impl.rs

@@ -24,20 +24,26 @@ use common_error::ext::BoxedError;
 use common_meta::error::{ExternalSnafu, Result, UnexpectedSnafu};
 use common_meta::node_manager::Flownode;
 use common_telemetry::{debug, trace};
+use datatypes::value::Value;
 use itertools::Itertools;
-use snafu::{OptionExt, ResultExt};
+use snafu::{IntoError, OptionExt, ResultExt};
 use store_api::storage::RegionId;
 
 use crate::adapter::{CreateFlowArgs, FlowWorkerManager};
-use crate::error::InternalSnafu;
+use crate::error::{CreateFlowSnafu, InsertIntoFlowSnafu, InternalSnafu};
 use crate::metrics::METRIC_FLOW_TASK_COUNT;
 use crate::repr::{self, DiffRow};
 
-fn to_meta_err(err: crate::error::Error) -> common_meta::error::Error {
-    // TODO(discord9): refactor this
-    Err::<(), _>(BoxedError::new(err))
-        .with_context(|_| ExternalSnafu)
-        .unwrap_err()
+/// return a function to convert `crate::error::Error` to `common_meta::error::Error`
+fn to_meta_err(
+    location: snafu::Location,
+) -> impl FnOnce(crate::error::Error) -> common_meta::error::Error {
+    move |err: crate::error::Error| -> common_meta::error::Error {
+        common_meta::error::Error::External {
+            location,
+            source: BoxedError::new(err),
+        }
+    }
 }
 
 #[async_trait::async_trait]
@@ -74,11 +80,16 @@ impl Flownode for FlowWorkerManager {
                     or_replace,
                     expire_after,
                     comment: Some(comment),
-                    sql,
+                    sql: sql.clone(),
                     flow_options,
                     query_ctx,
                 };
-                let ret = self.create_flow(args).await.map_err(to_meta_err)?;
+                let ret = self
+                    .create_flow(args)
+                    .await
+                    .map_err(BoxedError::new)
+                    .with_context(|_| CreateFlowSnafu { sql: sql.clone() })
+                    .map_err(to_meta_err(snafu::location!()))?;
                 METRIC_FLOW_TASK_COUNT.inc();
                 Ok(FlowResponse {
                     affected_flows: ret
@@ -93,7 +104,7 @@ impl Flownode for FlowWorkerManager {
             })) => {
                 self.remove_flow(flow_id.id as u64)
                     .await
-                    .map_err(to_meta_err)?;
+                    .map_err(to_meta_err(snafu::location!()))?;
                 METRIC_FLOW_TASK_COUNT.dec();
                 Ok(Default::default())
             }
@@ -111,9 +122,15 @@ impl Flownode for FlowWorkerManager {
                     .await
                     .flush_all_sender()
                     .await
-                    .map_err(to_meta_err)?;
-                let rows_send = self.run_available(true).await.map_err(to_meta_err)?;
-                let row = self.send_writeback_requests().await.map_err(to_meta_err)?;
+                    .map_err(to_meta_err(snafu::location!()))?;
+                let rows_send = self
+                    .run_available(true)
+                    .await
+                    .map_err(to_meta_err(snafu::location!()))?;
+                let row = self
+                    .send_writeback_requests()
+                    .await
+                    .map_err(to_meta_err(snafu::location!()))?;
 
                 debug!(
                     "Done to flush flow_id={:?} with {} input rows flushed, {} rows sended and {} output rows flushed",
@@ -137,7 +154,7 @@ impl Flownode for FlowWorkerManager {
     }
 
     async fn handle_inserts(&self, request: InsertRequests) -> Result<FlowResponse> {
-        // using try_read makesure two things:
+        // 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
         let _flush_lock = self.flush_lock.try_read();
@@ -153,17 +170,41 @@ impl Flownode for FlowWorkerManager {
             // TODO(discord9): reconsider time assignment mechanism
             let now = self.tick_manager.tick();
 
-            let fetch_order = {
+            let (table_types, fetch_order) = {
                 let ctx = self.node_context.read().await;
-                let table_col_names = ctx
-                    .table_repr
-                    .get_by_table_id(&table_id)
-                    .map(|r| r.1)
-                    .and_then(|id| ctx.schema.get(&id))
-                    .map(|desc| &desc.names)
-                    .context(UnexpectedSnafu {
-                        err_msg: format!("Table not found: {}", table_id),
-                    })?;
+
+                // TODO(discord9): also check schema version so that altered table can be reported
+                let table_schema = ctx
+                    .table_source
+                    .table_from_id(&table_id)
+                    .await
+                    .map_err(to_meta_err(snafu::location!()))?;
+                let default_vals = table_schema
+                    .default_values
+                    .iter()
+                    .zip(table_schema.relation_desc.typ().column_types.iter())
+                    .map(|(v, ty)| {
+                        v.as_ref().and_then(|v| {
+                            match v.create_default(ty.scalar_type(), ty.nullable()) {
+                                Ok(v) => Some(v),
+                                Err(err) => {
+                                    common_telemetry::error!(err; "Failed to create default value");
+                                    None
+                                }
+                            }
+                        })
+                    })
+                    .collect_vec();
+
+                let table_types = table_schema
+                    .relation_desc
+                    .typ()
+                    .column_types
+                    .clone()
+                    .into_iter()
+                    .map(|t| t.scalar_type)
+                    .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 {
@@ -180,36 +221,80 @@ impl Flownode for FlowWorkerManager {
                         .enumerate()
                         .map(|(i, name)| (&name.column_name, i)),
                 );
-                let fetch_order: Vec<usize> = table_col_names
+
+                let fetch_order: Vec<FetchFromRow> = table_col_names
                     .iter()
-                    .map(|names| {
-                        name_to_col.get(names).copied().context(UnexpectedSnafu {
-                            err_msg: format!("Column not found: {}", names),
-                        })
+                    .zip(default_vals.into_iter())
+                    .map(|(col_name, col_default_val)| {
+                        name_to_col
+                            .get(col_name)
+                            .copied()
+                            .map(FetchFromRow::Idx)
+                            .or_else(|| col_default_val.clone().map(FetchFromRow::Default))
+                            .with_context(|| UnexpectedSnafu {
+                                err_msg: format!(
+                                    "Column not found: {}, default_value: {:?}",
+                                    col_name, col_default_val
+                                ),
+                            })
                     })
                     .try_collect()?;
-                if !fetch_order.iter().enumerate().all(|(i, &v)| i == v) {
-                    trace!("Reordering columns: {:?}", fetch_order)
-                }
-                fetch_order
+
+                trace!("Reordering columns: {:?}", fetch_order);
+                (table_types, fetch_order)
             };
 
+            // TODO(discord9): use column instead of row
             let rows: Vec<DiffRow> = rows_proto
                 .into_iter()
                 .map(|r| {
                     let r = repr::Row::from(r);
-                    let reordered = fetch_order
-                        .iter()
-                        .map(|&i| r.inner[i].clone())
-                        .collect_vec();
+                    let reordered = fetch_order.iter().map(|i| i.fetch(&r)).collect_vec();
                     repr::Row::new(reordered)
                 })
                 .map(|r| (r, now, 1))
                 .collect_vec();
-            self.handle_write_request(region_id.into(), rows)
+            if let Err(err) = self
+                .handle_write_request(region_id.into(), rows, &table_types)
                 .await
-                .map_err(to_meta_err)?;
+            {
+                let err = BoxedError::new(err);
+                let flow_ids = self
+                    .node_context
+                    .read()
+                    .await
+                    .get_flow_ids(table_id)
+                    .into_iter()
+                    .flatten()
+                    .cloned()
+                    .collect_vec();
+                let err = InsertIntoFlowSnafu {
+                    region_id,
+                    flow_ids,
+                }
+                .into_error(err);
+                common_telemetry::error!(err; "Failed to handle write request");
+                let err = to_meta_err(snafu::location!())(err);
+                return Err(err);
+            }
         }
         Ok(Default::default())
     }
 }
+
+/// Simple helper enum for fetching value from row with default value
+#[derive(Debug, Clone)]
+enum FetchFromRow {
+    Idx(usize),
+    Default(Value),
+}
+
+impl FetchFromRow {
+    /// Panic if idx is out of bound
+    fn fetch(&self, row: &repr::Row) -> Value {
+        match self {
+            FetchFromRow::Idx(idx) => row.get(*idx).unwrap().clone(),
+            FetchFromRow::Default(v) => v.clone(),
+        }
+    }
+}