test!:Update config.yml

.github/ISSUE_TEMPLATE/config.yml

@@ -2,7 +2,7 @@ blank_issues_enabled: false
 contact_links:
   - name: Greptime Community Slack
     url: https://greptime.com/slack
-    about: Get free help from the Greptime community
+    about: Get free help from the Greptime community.
   - name: Greptime Community Discussion
     url: https://github.com/greptimeTeam/discussions
-    about: Get free help from the Greptime community
+    about: Get free help from the Greptime community.

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

@@ -52,7 +52,7 @@ runs:
       uses: ./.github/actions/build-greptime-binary
       with:
         base-image: ubuntu
-        features: servers/dashboard
+        features: servers/dashboard,pg_kvbackend,mysql_kvbackend
         cargo-profile: ${{ inputs.cargo-profile }}
         artifacts-dir: greptime-linux-${{ inputs.arch }}-${{ inputs.version }}
         version: ${{ inputs.version }}
@@ -70,7 +70,7 @@ runs:
       if: ${{ inputs.arch == 'amd64' && inputs.dev-mode == 'false' }} # Builds greptime for centos if the host machine is amd64.
       with:
         base-image: centos
-        features: servers/dashboard
+        features: servers/dashboard,pg_kvbackend,mysql_kvbackend
         cargo-profile: ${{ inputs.cargo-profile }}
         artifacts-dir: greptime-linux-${{ inputs.arch }}-centos-${{ inputs.version }}
         version: ${{ inputs.version }}

.github/scripts/create-version.sh

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

.github/workflows/dev-build.yml

@@ -55,11 +55,6 @@ on:
         description: Build and push images to DockerHub and ACR
         required: false
         default: true
-      upload_artifacts_to_s3:
-        type: boolean
-        description: Whether upload artifacts to s3
-        required: false
-        default: false
       cargo_profile:
         type: choice
         description: The cargo profile to use in building GreptimeDB.
@@ -243,7 +238,7 @@ jobs:
           version: ${{ needs.allocate-runners.outputs.version }}
           push-latest-tag: false # Don't push the latest tag to registry.
           dev-mode: true # Only build the standard images.
-
+          
       - name: Echo Docker image tag to step summary
         run: |
           echo "## Docker Image Tag" >> $GITHUB_STEP_SUMMARY
@@ -286,7 +281,7 @@ jobs:
           aws-cn-access-key-id: ${{ secrets.AWS_CN_ACCESS_KEY_ID }}
           aws-cn-secret-access-key: ${{ secrets.AWS_CN_SECRET_ACCESS_KEY }}
           aws-cn-region: ${{ vars.AWS_RELEASE_BUCKET_REGION }}
-          upload-to-s3: ${{ inputs.upload_artifacts_to_s3 }}
+          upload-to-s3: false
           dev-mode: true                     # Only build the standard images(exclude centos images).
           push-latest-tag: false             # Don't push the latest tag to registry.
           update-version-info: false         # Don't update the version info in S3.

.github/workflows/develop.yml

@@ -195,7 +195,6 @@ jobs:
     runs-on: ubuntu-latest
     timeout-minutes: 60
     strategy:
-      fail-fast: false
       matrix:
         target: [ "unstable_fuzz_create_table_standalone" ]
     steps:
@@ -300,7 +299,6 @@ jobs:
     needs:  build-greptime-ci
     timeout-minutes: 60
     strategy:
-      fail-fast: false
       matrix:
         target: [ "fuzz_create_table", "fuzz_alter_table", "fuzz_create_database", "fuzz_create_logical_table", "fuzz_alter_logical_table", "fuzz_insert", "fuzz_insert_logical_table" ]
         mode:
@@ -433,7 +431,6 @@ jobs:
     needs:  build-greptime-ci
     timeout-minutes: 60
     strategy:
-      fail-fast: false
       matrix:
         target: ["fuzz_migrate_mito_regions", "fuzz_migrate_metric_regions", "fuzz_failover_mito_regions", "fuzz_failover_metric_regions"]
         mode:
@@ -581,7 +578,6 @@ jobs:
     needs: build
     runs-on: ${{ matrix.os }}
     strategy:
-      fail-fast: false
       matrix:
         os: [ ubuntu-latest ]
         mode:

.github/workflows/nightly-ci.yml

@@ -124,7 +124,9 @@ jobs:
           fetch-depth: 0
           persist-credentials: false
       - uses: cachix/install-nix-action@v31
-      - run: nix develop --command cargo check --bin greptime
+        with:
+          nix_path: nixpkgs=channel:nixos-24.11
+      - run: nix develop --command cargo build --bin greptime
         env:
           CARGO_BUILD_RUSTFLAGS: "-C link-arg=-fuse-ld=mold"
 

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

@@ -16,7 +16,6 @@ jobs:
     runs-on: ubuntu-latest
     permissions:
       pull-requests: write  # Add permissions to modify PRs
-      issues: write
     timeout-minutes: 10
     steps:
       - uses: actions/checkout@v4

CONTRIBUTING.md

@@ -108,7 +108,7 @@ of what you were trying to do and what went wrong. You can also reach for help i
 The core team will be thrilled if you would like to participate in any way you like. When you are stuck, try to ask for help by filing an issue, with a detailed description of what you were trying to do and what went wrong. If you have any questions or if you would like to get involved in our community, please check out:
 
 - [GreptimeDB Community Slack](https://greptime.com/slack)
-- [GreptimeDB GitHub Discussions](https://github.com/GreptimeTeam/greptimedb/discussions)
+- [GreptimeDB Github Discussions](https://github.com/GreptimeTeam/greptimedb/discussions)
 
 Also, see some extra GreptimeDB content:
 

Cargo.lock

@@ -1852,9 +1852,8 @@ dependencies = [
  "futures",
  "humantime",
  "meta-client",
- "meta-srv",
  "nu-ansi-term",
- "object-store",
+ "opendal 0.51.2",
  "query",
  "rand 0.9.0",
  "reqwest",
@@ -1890,7 +1889,6 @@ dependencies = [
  "common-query",
  "common-recordbatch",
  "common-telemetry",
- "datatypes",
  "enum_dispatch",
  "futures",
  "futures-util",
@@ -2225,7 +2223,6 @@ version = "0.15.0"
 dependencies = [
  "api",
  "arrow-flight",
- "bytes",
  "common-base",
  "common-error",
  "common-macro",
@@ -2323,7 +2320,6 @@ dependencies = [
  "common-query",
  "common-recordbatch",
  "common-telemetry",
- "common-test-util",
  "common-time",
  "common-wal",
  "common-workload",
@@ -2334,7 +2330,6 @@ dependencies = [
  "deadpool-postgres",
  "derive_builder 0.20.1",
  "etcd-client",
- "flexbuffers",
  "futures",
  "futures-util",
  "hex",
@@ -2343,7 +2338,6 @@ dependencies = [
  "itertools 0.14.0",
  "lazy_static",
  "moka",
- "object-store",
  "prometheus",
  "prost 0.13.5",
  "rand 0.9.0",
@@ -2542,7 +2536,6 @@ name = "common-test-util"
 version = "0.15.0"
 dependencies = [
  "client",
- "common-grpc",
  "common-query",
  "common-recordbatch",
  "once_cell",
@@ -4265,19 +4258,6 @@ dependencies = [
  "miniz_oxide",
 ]
 
-[[package]]
-name = "flexbuffers"
-version = "25.2.10"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "935627e7bc8f083035d9faad09ffaed9128f73fb1f74a8798f115749c43378e8"
-dependencies = [
- "bitflags 1.3.2",
- "byteorder",
- "num_enum 0.5.11",
- "serde",
- "serde_derive",
-]
-
 [[package]]
 name = "float-cmp"
 version = "0.10.0"
@@ -4372,7 +4352,6 @@ dependencies = [
  "session",
  "smallvec",
  "snafu 0.8.5",
- "sql",
  "store-api",
  "strum 0.27.1",
  "substrait 0.15.0",
@@ -4876,7 +4855,7 @@ dependencies = [
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=2dca1dc67862d7b410838aef81232274c019b3f6#2dca1dc67862d7b410838aef81232274c019b3f6"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=7668a882d57ca6a2333146e0574b8f0c9d5008ae#7668a882d57ca6a2333146e0574b8f0c9d5008ae"
 dependencies = [
  "prost 0.13.5",
  "serde",
@@ -6976,7 +6955,6 @@ dependencies = [
  "common-decimal",
  "common-error",
  "common-function",
- "common-grpc",
  "common-macro",
  "common-meta",
  "common-query",
@@ -7594,34 +7572,13 @@ dependencies = [
  "libc",
 ]
 
-[[package]]
-name = "num_enum"
-version = "0.5.11"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1f646caf906c20226733ed5b1374287eb97e3c2a5c227ce668c1f2ce20ae57c9"
-dependencies = [
- "num_enum_derive 0.5.11",
-]
-
 [[package]]
 name = "num_enum"
 version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4e613fc340b2220f734a8595782c551f1250e969d87d3be1ae0579e8d4065179"
 dependencies = [
- "num_enum_derive 0.7.3",
-]
-
-[[package]]
-name = "num_enum_derive"
-version = "0.5.11"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dcbff9bc912032c62bf65ef1d5aea88983b420f4f839db1e9b0c281a25c9c799"
-dependencies = [
- "proc-macro-crate 1.3.1",
- "proc-macro2",
- "quote",
- "syn 1.0.109",
+ "num_enum_derive",
 ]
 
 [[package]]
@@ -7675,7 +7632,7 @@ dependencies = [
  "lazy_static",
  "md5",
  "moka",
- "opendal",
+ "opendal 0.52.0",
  "prometheus",
  "tokio",
  "uuid",
@@ -7714,7 +7671,7 @@ dependencies = [
  "futures",
  "futures-util",
  "object_store",
- "opendal",
+ "opendal 0.52.0",
  "pin-project",
  "tokio",
 ]
@@ -7740,6 +7697,35 @@ version = "11.1.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b410bbe7e14ab526a0e86877eb47c6996a2bd7746f027ba551028c925390e4e9"
 
+[[package]]
+name = "opendal"
+version = "0.51.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5b1063ea459fa9e94584115743b06330f437902dd1d9f692b863ef1875a20548"
+dependencies = [
+ "anyhow",
+ "async-trait",
+ "backon",
+ "base64 0.22.1",
+ "bytes",
+ "chrono",
+ "crc32c",
+ "futures",
+ "getrandom 0.2.15",
+ "http 1.1.0",
+ "log",
+ "md-5",
+ "once_cell",
+ "percent-encoding",
+ "quick-xml 0.36.2",
+ "reqsign",
+ "reqwest",
+ "serde",
+ "serde_json",
+ "tokio",
+ "uuid",
+]
+
 [[package]]
 name = "opendal"
 version = "0.52.0"
@@ -8091,7 +8077,7 @@ dependencies = [
  "arrow 53.4.1",
  "arrow-ipc 53.4.1",
  "lazy_static",
- "num_enum 0.7.3",
+ "num_enum",
  "opentelemetry-proto 0.27.0",
  "paste",
  "prost 0.13.5",
@@ -8428,9 +8414,9 @@ dependencies = [
 
 [[package]]
 name = "pgwire"
-version = "0.30.1"
+version = "0.29.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ec79ee18e6cafde8698885646780b967ecc905120798b8359dd0da64f9688e89"
+checksum = "a4e6fcdc2ae2173ef8ee1005b6e46453d45195ac3d97caac0db7ecf64ab4aa85"
 dependencies = [
  "async-trait",
  "bytes",

Cargo.toml

@@ -132,7 +132,7 @@ etcd-client = "0.14"
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "2dca1dc67862d7b410838aef81232274c019b3f6" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "7668a882d57ca6a2333146e0574b8f0c9d5008ae" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"

config/config.md

@@ -27,7 +27,6 @@
 | `http.body_limit` | String | `64MB` | HTTP request body limit.<br/>The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.<br/>Set to 0 to disable limit. |
 | `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default<br/>This allows browser to access http APIs without CORS restrictions |
 | `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. |
-| `http.prom_validation_mode` | String | `strict` | Whether to enable validation for Prometheus remote write requests.<br/>Available options:<br/>- strict: deny invalid UTF-8 strings (default).<br/>- lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).<br/>- unchecked: do not valid strings. |
 | `grpc` | -- | -- | The gRPC server options. |
 | `grpc.bind_addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. |
 | `grpc.runtime_size` | Integer | `8` | The number of server worker threads. |
@@ -227,7 +226,6 @@
 | `http.body_limit` | String | `64MB` | HTTP request body limit.<br/>The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.<br/>Set to 0 to disable limit. |
 | `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default<br/>This allows browser to access http APIs without CORS restrictions |
 | `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. |
-| `http.prom_validation_mode` | String | `strict` | Whether to enable validation for Prometheus remote write requests.<br/>Available options:<br/>- strict: deny invalid UTF-8 strings (default).<br/>- lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).<br/>- unchecked: do not valid strings. |
 | `grpc` | -- | -- | The gRPC server options. |
 | `grpc.bind_addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. |
 | `grpc.server_addr` | String | `127.0.0.1:4001` | The address advertised to the metasrv, and used for connections from outside the host.<br/>If left empty or unset, the server will automatically use the IP address of the first network interface<br/>on the host, with the same port number as the one specified in `grpc.bind_addr`. |
@@ -331,10 +329,6 @@
 | `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. |
-| `http` | -- | -- | The HTTP server options. |
-| `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. |
-| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout. |
-| `http.body_limit` | String | `64MB` | HTTP request body limit.<br/>The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.<br/>Set to 0 to disable limit. |
 | `procedure` | -- | -- | Procedure storage options. |
 | `procedure.max_retry_times` | Integer | `12` | Procedure max retry time. |
 | `procedure.retry_delay` | String | `500ms` | Initial retry delay of procedures, increases exponentially |

config/frontend.example.toml

@@ -37,12 +37,6 @@ enable_cors = true
 ## Customize allowed origins for HTTP CORS.
 ## @toml2docs:none-default
 cors_allowed_origins = ["https://example.com"]
-## Whether to enable validation for Prometheus remote write requests.
-## Available options:
-## - strict: deny invalid UTF-8 strings (default).
-## - lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).
-## - unchecked: do not valid strings.
-prom_validation_mode = "strict"
 
 ## The gRPC server options.
 [grpc]

config/metasrv.example.toml

@@ -67,17 +67,6 @@ node_max_idle_time = "24hours"
 ## The number of threads to execute the runtime for global write operations.
 #+ compact_rt_size = 4
 
-## The HTTP server options.
-[http]
-## The address to bind the HTTP server.
-addr = "127.0.0.1:4000"
-## HTTP request timeout. Set to 0 to disable timeout.
-timeout = "0s"
-## HTTP request body limit.
-## The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.
-## Set to 0 to disable limit.
-body_limit = "64MB"
-
 ## Procedure storage options.
 [procedure]
 

config/standalone.example.toml

@@ -43,13 +43,6 @@ enable_cors = true
 ## @toml2docs:none-default
 cors_allowed_origins = ["https://example.com"]
 
-## Whether to enable validation for Prometheus remote write requests.
-## Available options:
-## - strict: deny invalid UTF-8 strings (default).
-## - lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).
-## - unchecked: do not valid strings.
-prom_validation_mode = "strict"
-
 ## The gRPC server options.
 [grpc]
 ## The address to bind the gRPC server.

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

@@ -1,6 +1,6 @@
 # Profile memory usage of GreptimeDB
 
-This crate provides an easy approach to dump memory profiling info. A set of ready to use scripts is provided in [docs/how-to/memory-profile-scripts](./memory-profile-scripts/scripts).
+This crate provides an easy approach to dump memory profiling info. A set of ready to use scripts is provided in [docs/how-to/memory-profile-scripts](docs/how-to/memory-profile-scripts).
 
 ## Prerequisites
 ### jemalloc

flake.lock

@@ -41,16 +41,16 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1748162331,
-        "narHash": "sha256-rqc2RKYTxP3tbjA+PB3VMRQNnjesrT0pEofXQTrMsS8=",
+        "lastModified": 1745487689,
+        "narHash": "sha256-FQoi3R0NjQeBAsEOo49b5tbDPcJSMWc3QhhaIi9eddw=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "7c43f080a7f28b2774f3b3f43234ca11661bf334",
+        "rev": "5630cf13cceac06cefe9fc607e8dfa8fb342dde3",
         "type": "github"
       },
       "original": {
         "owner": "NixOS",
-        "ref": "nixos-25.05",
+        "ref": "nixos-24.11",
         "repo": "nixpkgs",
         "type": "github"
       }

flake.nix

@@ -2,7 +2,7 @@
   description = "Development environment flake";
 
   inputs = {
-    nixpkgs.url = "github:NixOS/nixpkgs/nixos-25.05";
+    nixpkgs.url = "github:NixOS/nixpkgs/nixos-24.11";
     fenix = {
       url = "github:nix-community/fenix";
       inputs.nixpkgs.follows = "nixpkgs";
@@ -51,7 +51,6 @@
           ];
 
           LD_LIBRARY_PATH = pkgs.lib.makeLibraryPath buildInputs;
-          NIX_HARDENING_ENABLE = "";
         };
       });
 }

grafana/dashboards/metrics/cluster/dashboard.json

@@ -25,7 +25,7 @@
   "editable": true,
   "fiscalYearStartMonth": 0,
   "graphTooltip": 1,
-  "id": 7,
+  "id": 5,
   "links": [],
   "panels": [
     {
@@ -4476,7 +4476,7 @@
             "axisPlacement": "auto",
             "barAlignment": 0,
             "barWidthFactor": 0.6,
-            "drawStyle": "points",
+            "drawStyle": "line",
             "fillOpacity": 0,
             "gradientMode": "none",
             "hideFrom": {
@@ -4553,22 +4553,9 @@
           "legendFormat": "[{{instance}}]-[{{pod}}]-[{{stage}}]-p99",
           "range": true,
           "refId": "A"
-        },
-        {
-          "datasource": {
-            "type": "prometheus",
-            "uid": "${metrics}"
-          },
-          "editorMode": "code",
-          "expr": "sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_sum{instance=~\"$datanode\"}[$__rate_interval]))/sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_count{instance=~\"$datanode\"}[$__rate_interval]))",
-          "hide": false,
-          "instant": false,
-          "legendFormat": "[{{instance}}]-[{{pod}}]-[{{stage}}]-avg",
-          "range": true,
-          "refId": "B"
         }
       ],
-      "title": "Compaction Elapsed Time per Instance by Stage",
+      "title": "Compaction P99 per Instance by Stage",
       "type": "timeseries"
     },
     {
@@ -5559,131 +5546,13 @@
       "title": "Region Worker Handle Bulk Insert Requests",
       "type": "timeseries"
     },
-    {
-      "datasource": {
-        "type": "prometheus",
-        "uid": "${metrics}"
-      },
-      "description": "Per-stage elapsed time for region worker to decode requests.",
-      "fieldConfig": {
-        "defaults": {
-          "color": {
-            "mode": "palette-classic"
-          },
-          "custom": {
-            "axisBorderShow": false,
-            "axisCenteredZero": false,
-            "axisColorMode": "text",
-            "axisLabel": "",
-            "axisPlacement": "auto",
-            "barAlignment": 0,
-            "barWidthFactor": 0.6,
-            "drawStyle": "points",
-            "fillOpacity": 0,
-            "gradientMode": "none",
-            "hideFrom": {
-              "legend": false,
-              "tooltip": false,
-              "viz": false
-            },
-            "insertNulls": false,
-            "lineInterpolation": "linear",
-            "lineWidth": 1,
-            "pointSize": 5,
-            "scaleDistribution": {
-              "type": "linear"
-            },
-            "showPoints": "auto",
-            "spanNulls": false,
-            "stacking": {
-              "group": "A",
-              "mode": "none"
-            },
-            "thresholdsStyle": {
-              "mode": "off"
-            }
-          },
-          "mappings": [],
-          "thresholds": {
-            "mode": "absolute",
-            "steps": [
-              {
-                "color": "green"
-              },
-              {
-                "color": "red",
-                "value": 80
-              }
-            ]
-          },
-          "unit": "s"
-        },
-        "overrides": []
-      },
-      "gridPos": {
-        "h": 8,
-        "w": 12,
-        "x": 12,
-        "y": 117
-      },
-      "id": 338,
-      "options": {
-        "legend": {
-          "calcs": [
-            "lastNotNull"
-          ],
-          "displayMode": "table",
-          "placement": "bottom",
-          "showLegend": true
-        },
-        "tooltip": {
-          "hideZeros": false,
-          "mode": "single",
-          "sort": "none"
-        }
-      },
-      "pluginVersion": "12.0.0",
-      "targets": [
-        {
-          "datasource": {
-            "type": "prometheus",
-            "uid": "${metrics}"
-          },
-          "disableTextWrap": false,
-          "editorMode": "builder",
-          "expr": "histogram_quantile(0.95, sum by(le, instance, stage, pod) (rate(greptime_datanode_convert_region_request_bucket[$__rate_interval])))",
-          "fullMetaSearch": false,
-          "includeNullMetadata": true,
-          "instant": false,
-          "legendFormat": "[{{instance}}]-[{{pod}}]-[{{stage}}]-P95",
-          "range": true,
-          "refId": "A",
-          "useBackend": false
-        },
-        {
-          "datasource": {
-            "type": "prometheus",
-            "uid": "${metrics}"
-          },
-          "editorMode": "code",
-          "expr": "sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_sum[$__rate_interval]))/sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_count[$__rate_interval]))",
-          "hide": false,
-          "instant": false,
-          "legendFormat": "[{{instance}}]-[{{pod}}]-[{{stage}}]-AVG",
-          "range": true,
-          "refId": "B"
-        }
-      ],
-      "title": "Region Worker Convert Requests",
-      "type": "timeseries"
-    },
     {
       "collapsed": true,
       "gridPos": {
         "h": 1,
         "w": 24,
         "x": 0,
-        "y": 125
+        "y": 117
       },
       "id": 313,
       "panels": [
@@ -6813,7 +6682,7 @@
         "h": 1,
         "w": 24,
         "x": 0,
-        "y": 126
+        "y": 118
       },
       "id": 324,
       "panels": [
@@ -7110,7 +6979,7 @@
         "h": 1,
         "w": 24,
         "x": 0,
-        "y": 127
+        "y": 119
       },
       "id": 328,
       "panels": [
@@ -7758,5 +7627,6 @@
   "timezone": "",
   "title": "GreptimeDB",
   "uid": "dejf3k5e7g2kgb",
-  "version": 3
+  "version": 3,
+  "weekStart": ""
 }

grafana/dashboards/metrics/cluster/dashboard.md

@@ -60,7 +60,7 @@
 | Read Stage P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_read_stage_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))` | `timeseries` | Read Stage P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]` |
 | Write Stage P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_write_stage_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))` | `timeseries` | Write Stage P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]` |
 | Compaction OPS per Instance | `sum by(instance, pod) (rate(greptime_mito_compaction_total_elapsed_count{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | Compaction OPS per Instance. | `prometheus` | `ops` | `[{{ instance }}]-[{{pod}}]` |
-| Compaction Elapsed Time per Instance by Stage | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))`<br/>`sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_sum{instance=~"$datanode"}[$__rate_interval]))/sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_count{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | Compaction latency by stage | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-p99` |
+| Compaction P99 per Instance by Stage | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))` | `timeseries` | Compaction latency by stage | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-p99` |
 | Compaction P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le,stage) (rate(greptime_mito_compaction_total_elapsed_bucket{instance=~"$datanode"}[$__rate_interval])))` | `timeseries` | Compaction P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-compaction` |
 | WAL write size | `histogram_quantile(0.95, sum by(le,instance, pod) (rate(raft_engine_write_size_bucket[$__rate_interval])))`<br/>`histogram_quantile(0.99, sum by(le,instance,pod) (rate(raft_engine_write_size_bucket[$__rate_interval])))`<br/>`sum by (instance, pod)(rate(raft_engine_write_size_sum[$__rate_interval]))` | `timeseries` | Write-ahead logs write size as bytes. This chart includes stats of p95 and p99 size by instance, total WAL write rate. | `prometheus` | `bytes` | `[{{instance}}]-[{{pod}}]-req-size-p95` |
 | Cached Bytes per Instance | `greptime_mito_cache_bytes{instance=~"$datanode"}` | `timeseries` | Cached Bytes per Instance. | `prometheus` | `decbytes` | `[{{instance}}]-[{{pod}}]-[{{type}}]` |
@@ -70,7 +70,6 @@
 | Inflight Flush | `greptime_mito_inflight_flush_count` | `timeseries` | Ongoing flush task count | `prometheus` | `none` | `[{{instance}}]-[{{pod}}]` |
 | Compaction Input/Output Bytes | `sum by(instance, pod) (greptime_mito_compaction_input_bytes)`<br/>`sum by(instance, pod) (greptime_mito_compaction_output_bytes)` | `timeseries` | Compaction oinput output bytes | `prometheus` | `bytes` | `[{{instance}}]-[{{pod}}]-input` |
 | Region Worker Handle Bulk Insert Requests | `histogram_quantile(0.95, sum by(le,instance, stage, pod) (rate(greptime_region_worker_handle_write_bucket[$__rate_interval])))`<br/>`sum by(le,instance, stage, pod) (rate(greptime_region_worker_handle_write_sum[$__rate_interval]))/sum by(le,instance, stage, pod) (rate(greptime_region_worker_handle_write_count[$__rate_interval]))` | `timeseries` | Per-stage elapsed time for region worker to handle bulk insert region requests. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-P95` |
-| Region Worker Convert Requests | `histogram_quantile(0.95, sum by(le, instance, stage, pod) (rate(greptime_datanode_convert_region_request_bucket[$__rate_interval])))`<br/>`sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_sum[$__rate_interval]))/sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_count[$__rate_interval]))` | `timeseries` | Per-stage elapsed time for region worker to decode requests. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-P95` |
 # OpenDAL
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |

grafana/dashboards/metrics/cluster/dashboard.yaml

@@ -487,7 +487,7 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{ instance }}]-[{{pod}}]'
-        - title: Compaction Elapsed Time per Instance by Stage
+        - title: Compaction P99 per Instance by Stage
           type: timeseries
           description: Compaction latency by stage
           unit: s
@@ -497,11 +497,6 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-p99'
-            - expr: sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_sum{instance=~"$datanode"}[$__rate_interval]))/sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_count{instance=~"$datanode"}[$__rate_interval]))
-              datasource:
-                type: prometheus
-                uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-avg'
         - title: Compaction P99 per Instance
           type: timeseries
           description: Compaction P99 per Instance.
@@ -612,21 +607,6 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-AVG'
-        - title: Region Worker Convert Requests
-          type: timeseries
-          description: Per-stage elapsed time for region worker to decode requests.
-          unit: s
-          queries:
-            - expr: histogram_quantile(0.95, sum by(le, instance, stage, pod) (rate(greptime_datanode_convert_region_request_bucket[$__rate_interval])))
-              datasource:
-                type: prometheus
-                uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-P95'
-            - expr: sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_sum[$__rate_interval]))/sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_count[$__rate_interval]))
-              datasource:
-                type: prometheus
-                uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-AVG'
     - title: OpenDAL
       panels:
         - title: QPS per Instance

grafana/dashboards/metrics/standalone/dashboard.json

@@ -25,7 +25,7 @@
   "editable": true,
   "fiscalYearStartMonth": 0,
   "graphTooltip": 1,
-  "id": 7,
+  "id": 5,
   "links": [],
   "panels": [
     {
@@ -4476,7 +4476,7 @@
             "axisPlacement": "auto",
             "barAlignment": 0,
             "barWidthFactor": 0.6,
-            "drawStyle": "points",
+            "drawStyle": "line",
             "fillOpacity": 0,
             "gradientMode": "none",
             "hideFrom": {
@@ -4553,22 +4553,9 @@
           "legendFormat": "[{{instance}}]-[{{pod}}]-[{{stage}}]-p99",
           "range": true,
           "refId": "A"
-        },
-        {
-          "datasource": {
-            "type": "prometheus",
-            "uid": "${metrics}"
-          },
-          "editorMode": "code",
-          "expr": "sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_sum{}[$__rate_interval]))/sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_count{}[$__rate_interval]))",
-          "hide": false,
-          "instant": false,
-          "legendFormat": "[{{instance}}]-[{{pod}}]-[{{stage}}]-avg",
-          "range": true,
-          "refId": "B"
         }
       ],
-      "title": "Compaction Elapsed Time per Instance by Stage",
+      "title": "Compaction P99 per Instance by Stage",
       "type": "timeseries"
     },
     {
@@ -5559,131 +5546,13 @@
       "title": "Region Worker Handle Bulk Insert Requests",
       "type": "timeseries"
     },
-    {
-      "datasource": {
-        "type": "prometheus",
-        "uid": "${metrics}"
-      },
-      "description": "Per-stage elapsed time for region worker to decode requests.",
-      "fieldConfig": {
-        "defaults": {
-          "color": {
-            "mode": "palette-classic"
-          },
-          "custom": {
-            "axisBorderShow": false,
-            "axisCenteredZero": false,
-            "axisColorMode": "text",
-            "axisLabel": "",
-            "axisPlacement": "auto",
-            "barAlignment": 0,
-            "barWidthFactor": 0.6,
-            "drawStyle": "points",
-            "fillOpacity": 0,
-            "gradientMode": "none",
-            "hideFrom": {
-              "legend": false,
-              "tooltip": false,
-              "viz": false
-            },
-            "insertNulls": false,
-            "lineInterpolation": "linear",
-            "lineWidth": 1,
-            "pointSize": 5,
-            "scaleDistribution": {
-              "type": "linear"
-            },
-            "showPoints": "auto",
-            "spanNulls": false,
-            "stacking": {
-              "group": "A",
-              "mode": "none"
-            },
-            "thresholdsStyle": {
-              "mode": "off"
-            }
-          },
-          "mappings": [],
-          "thresholds": {
-            "mode": "absolute",
-            "steps": [
-              {
-                "color": "green"
-              },
-              {
-                "color": "red",
-                "value": 80
-              }
-            ]
-          },
-          "unit": "s"
-        },
-        "overrides": []
-      },
-      "gridPos": {
-        "h": 8,
-        "w": 12,
-        "x": 12,
-        "y": 117
-      },
-      "id": 338,
-      "options": {
-        "legend": {
-          "calcs": [
-            "lastNotNull"
-          ],
-          "displayMode": "table",
-          "placement": "bottom",
-          "showLegend": true
-        },
-        "tooltip": {
-          "hideZeros": false,
-          "mode": "single",
-          "sort": "none"
-        }
-      },
-      "pluginVersion": "12.0.0",
-      "targets": [
-        {
-          "datasource": {
-            "type": "prometheus",
-            "uid": "${metrics}"
-          },
-          "disableTextWrap": false,
-          "editorMode": "builder",
-          "expr": "histogram_quantile(0.95, sum by(le, instance, stage, pod) (rate(greptime_datanode_convert_region_request_bucket[$__rate_interval])))",
-          "fullMetaSearch": false,
-          "includeNullMetadata": true,
-          "instant": false,
-          "legendFormat": "[{{instance}}]-[{{pod}}]-[{{stage}}]-P95",
-          "range": true,
-          "refId": "A",
-          "useBackend": false
-        },
-        {
-          "datasource": {
-            "type": "prometheus",
-            "uid": "${metrics}"
-          },
-          "editorMode": "code",
-          "expr": "sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_sum[$__rate_interval]))/sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_count[$__rate_interval]))",
-          "hide": false,
-          "instant": false,
-          "legendFormat": "[{{instance}}]-[{{pod}}]-[{{stage}}]-AVG",
-          "range": true,
-          "refId": "B"
-        }
-      ],
-      "title": "Region Worker Convert Requests",
-      "type": "timeseries"
-    },
     {
       "collapsed": true,
       "gridPos": {
         "h": 1,
         "w": 24,
         "x": 0,
-        "y": 125
+        "y": 117
       },
       "id": 313,
       "panels": [
@@ -6813,7 +6682,7 @@
         "h": 1,
         "w": 24,
         "x": 0,
-        "y": 126
+        "y": 118
       },
       "id": 324,
       "panels": [
@@ -7110,7 +6979,7 @@
         "h": 1,
         "w": 24,
         "x": 0,
-        "y": 127
+        "y": 119
       },
       "id": 328,
       "panels": [
@@ -7758,5 +7627,6 @@
   "timezone": "",
   "title": "GreptimeDB",
   "uid": "dejf3k5e7g2kgb",
-  "version": 3
+  "version": 3,
+  "weekStart": ""
 }

grafana/dashboards/metrics/standalone/dashboard.md

@@ -60,7 +60,7 @@
 | Read Stage P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_read_stage_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | Read Stage P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]` |
 | Write Stage P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_write_stage_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | Write Stage P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]` |
 | Compaction OPS per Instance | `sum by(instance, pod) (rate(greptime_mito_compaction_total_elapsed_count{}[$__rate_interval]))` | `timeseries` | Compaction OPS per Instance. | `prometheus` | `ops` | `[{{ instance }}]-[{{pod}}]` |
-| Compaction Elapsed Time per Instance by Stage | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_bucket{}[$__rate_interval])))`<br/>`sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_sum{}[$__rate_interval]))/sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_count{}[$__rate_interval]))` | `timeseries` | Compaction latency by stage | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-p99` |
+| Compaction P99 per Instance by Stage | `histogram_quantile(0.99, sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | Compaction latency by stage | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-p99` |
 | Compaction P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le,stage) (rate(greptime_mito_compaction_total_elapsed_bucket{}[$__rate_interval])))` | `timeseries` | Compaction P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-compaction` |
 | WAL write size | `histogram_quantile(0.95, sum by(le,instance, pod) (rate(raft_engine_write_size_bucket[$__rate_interval])))`<br/>`histogram_quantile(0.99, sum by(le,instance,pod) (rate(raft_engine_write_size_bucket[$__rate_interval])))`<br/>`sum by (instance, pod)(rate(raft_engine_write_size_sum[$__rate_interval]))` | `timeseries` | Write-ahead logs write size as bytes. This chart includes stats of p95 and p99 size by instance, total WAL write rate. | `prometheus` | `bytes` | `[{{instance}}]-[{{pod}}]-req-size-p95` |
 | Cached Bytes per Instance | `greptime_mito_cache_bytes{}` | `timeseries` | Cached Bytes per Instance. | `prometheus` | `decbytes` | `[{{instance}}]-[{{pod}}]-[{{type}}]` |
@@ -70,7 +70,6 @@
 | Inflight Flush | `greptime_mito_inflight_flush_count` | `timeseries` | Ongoing flush task count | `prometheus` | `none` | `[{{instance}}]-[{{pod}}]` |
 | Compaction Input/Output Bytes | `sum by(instance, pod) (greptime_mito_compaction_input_bytes)`<br/>`sum by(instance, pod) (greptime_mito_compaction_output_bytes)` | `timeseries` | Compaction oinput output bytes | `prometheus` | `bytes` | `[{{instance}}]-[{{pod}}]-input` |
 | Region Worker Handle Bulk Insert Requests | `histogram_quantile(0.95, sum by(le,instance, stage, pod) (rate(greptime_region_worker_handle_write_bucket[$__rate_interval])))`<br/>`sum by(le,instance, stage, pod) (rate(greptime_region_worker_handle_write_sum[$__rate_interval]))/sum by(le,instance, stage, pod) (rate(greptime_region_worker_handle_write_count[$__rate_interval]))` | `timeseries` | Per-stage elapsed time for region worker to handle bulk insert region requests. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-P95` |
-| Region Worker Convert Requests | `histogram_quantile(0.95, sum by(le, instance, stage, pod) (rate(greptime_datanode_convert_region_request_bucket[$__rate_interval])))`<br/>`sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_sum[$__rate_interval]))/sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_count[$__rate_interval]))` | `timeseries` | Per-stage elapsed time for region worker to decode requests. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-P95` |
 # OpenDAL
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |

grafana/dashboards/metrics/standalone/dashboard.yaml

@@ -487,7 +487,7 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{ instance }}]-[{{pod}}]'
-        - title: Compaction Elapsed Time per Instance by Stage
+        - title: Compaction P99 per Instance by Stage
           type: timeseries
           description: Compaction latency by stage
           unit: s
@@ -497,11 +497,6 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-p99'
-            - expr: sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_sum{}[$__rate_interval]))/sum by(instance, pod, le, stage) (rate(greptime_mito_compaction_stage_elapsed_count{}[$__rate_interval]))
-              datasource:
-                type: prometheus
-                uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-avg'
         - title: Compaction P99 per Instance
           type: timeseries
           description: Compaction P99 per Instance.
@@ -612,21 +607,6 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-AVG'
-        - title: Region Worker Convert Requests
-          type: timeseries
-          description: Per-stage elapsed time for region worker to decode requests.
-          unit: s
-          queries:
-            - expr: histogram_quantile(0.95, sum by(le, instance, stage, pod) (rate(greptime_datanode_convert_region_request_bucket[$__rate_interval])))
-              datasource:
-                type: prometheus
-                uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-P95'
-            - expr: sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_sum[$__rate_interval]))/sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_count[$__rate_interval]))
-              datasource:
-                type: prometheus
-                uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-AVG'
     - title: OpenDAL
       panels:
         - title: QPS per Instance

grafana/scripts/gen-dashboards.sh

@@ -6,7 +6,7 @@ DAC_IMAGE=ghcr.io/zyy17/dac:20250423-522bd35
 
 remove_instance_filters() {
   # Remove the instance filters for the standalone dashboards.
-  sed -E 's/instance=~\\"(\$datanode|\$frontend|\$metasrv|\$flownode)\\",?//g' "$CLUSTER_DASHBOARD_DIR/dashboard.json" > "$STANDALONE_DASHBOARD_DIR/dashboard.json"
+  sed 's/instance=~\\"$datanode\\",//; s/instance=~\\"$datanode\\"//; s/instance=~\\"$frontend\\",//; s/instance=~\\"$frontend\\"//; s/instance=~\\"$metasrv\\",//; s/instance=~\\"$metasrv\\"//; s/instance=~\\"$flownode\\",//; s/instance=~\\"$flownode\\"//;' $CLUSTER_DASHBOARD_DIR/dashboard.json > $STANDALONE_DASHBOARD_DIR/dashboard.json
 }
 
 generate_intermediate_dashboards_and_docs() {

licenserc.toml

@@ -26,13 +26,6 @@ excludes = [
     "src/common/base/src/secrets.rs",
     "src/servers/src/repeated_field.rs",
     "src/servers/src/http/test_helpers.rs",
-    # enterprise
-    "src/common/meta/src/rpc/ddl/trigger.rs",
-    "src/operator/src/expr_helper/trigger.rs",
-    "src/sql/src/statements/create/trigger.rs",
-    "src/sql/src/statements/show/trigger.rs",
-    "src/sql/src/parsers/create_parser/trigger.rs",
-    "src/sql/src/parsers/show_parser/trigger.rs",
 ]
 
 [properties]

src/cli/Cargo.toml

@@ -5,12 +5,8 @@ edition.workspace = true
 license.workspace = true
 
 [features]
-default = [
-    "pg_kvbackend",
-    "mysql_kvbackend",
-]
-pg_kvbackend = ["common-meta/pg_kvbackend", "meta-srv/pg_kvbackend"]
-mysql_kvbackend = ["common-meta/mysql_kvbackend", "meta-srv/mysql_kvbackend"]
+pg_kvbackend = ["common-meta/pg_kvbackend"]
+mysql_kvbackend = ["common-meta/mysql_kvbackend"]
 
 [lints]
 workspace = true
@@ -47,9 +43,11 @@ etcd-client.workspace = true
 futures.workspace = true
 humantime.workspace = true
 meta-client.workspace = true
-meta-srv.workspace = true
 nu-ansi-term = "0.46"
-object-store.workspace = true
+opendal = { version = "0.51.1", features = [
+    "services-fs",
+    "services-s3",
+] }
 query.workspace = true
 rand.workspace = true
 reqwest.workspace = true

src/cli/src/error.rs

@@ -17,7 +17,6 @@ use std::any::Any;
 use common_error::ext::{BoxedError, ErrorExt};
 use common_error::status_code::StatusCode;
 use common_macro::stack_trace_debug;
-use object_store::Error as ObjectStoreError;
 use snafu::{Location, Snafu};
 
 #[derive(Snafu)]
@@ -226,7 +225,7 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
         #[snafu(source)]
-        error: ObjectStoreError,
+        error: opendal::Error,
     },
     #[snafu(display("S3 config need be set"))]
     S3ConfigNotSet {
@@ -238,12 +237,6 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
-    #[snafu(display("KV backend not set: {}", backend))]
-    KvBackendNotSet {
-        backend: String,
-        #[snafu(implicit)]
-        location: Location,
-    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -280,9 +273,8 @@ impl ErrorExt for Error {
 
             Error::Other { source, .. } => source.status_code(),
             Error::OpenDal { .. } => StatusCode::Internal,
-            Error::S3ConfigNotSet { .. }
-            | Error::OutputDirNotSet { .. }
-            | Error::KvBackendNotSet { .. } => StatusCode::InvalidArguments,
+            Error::S3ConfigNotSet { .. } => StatusCode::InvalidArguments,
+            Error::OutputDirNotSet { .. } => StatusCode::InvalidArguments,
 
             Error::BuildRuntime { source, .. } => source.status_code(),
 

src/cli/src/export.rs

@@ -21,8 +21,8 @@ use async_trait::async_trait;
 use clap::{Parser, ValueEnum};
 use common_error::ext::BoxedError;
 use common_telemetry::{debug, error, info};
-use object_store::layers::LoggingLayer;
-use object_store::{services, ObjectStore};
+use opendal::layers::LoggingLayer;
+use opendal::{services, Operator};
 use serde_json::Value;
 use snafu::{OptionExt, ResultExt};
 use tokio::sync::Semaphore;
@@ -470,7 +470,7 @@ impl Export {
         Ok(())
     }
 
-    async fn build_operator(&self) -> Result<ObjectStore> {
+    async fn build_operator(&self) -> Result<Operator> {
         if self.s3 {
             self.build_s3_operator().await
         } else {
@@ -479,11 +479,11 @@ impl Export {
     }
 
     /// build operator with preference for file system
-    async fn build_prefer_fs_operator(&self) -> Result<ObjectStore> {
+    async fn build_prefer_fs_operator(&self) -> Result<Operator> {
         // is under s3 mode and s3_ddl_dir is set, use it as root
         if self.s3 && self.s3_ddl_local_dir.is_some() {
             let root = self.s3_ddl_local_dir.as_ref().unwrap().clone();
-            let op = ObjectStore::new(services::Fs::default().root(&root))
+            let op = Operator::new(services::Fs::default().root(&root))
                 .context(OpenDalSnafu)?
                 .layer(LoggingLayer::default())
                 .finish();
@@ -495,7 +495,7 @@ impl Export {
         }
     }
 
-    async fn build_s3_operator(&self) -> Result<ObjectStore> {
+    async fn build_s3_operator(&self) -> Result<Operator> {
         let mut builder = services::S3::default().bucket(
             self.s3_bucket
                 .as_ref()
@@ -522,20 +522,20 @@ impl Export {
             builder = builder.secret_access_key(secret_key);
         }
 
-        let op = ObjectStore::new(builder)
+        let op = Operator::new(builder)
             .context(OpenDalSnafu)?
             .layer(LoggingLayer::default())
             .finish();
         Ok(op)
     }
 
-    async fn build_fs_operator(&self) -> Result<ObjectStore> {
+    async fn build_fs_operator(&self) -> Result<Operator> {
         let root = self
             .output_dir
             .as_ref()
             .context(OutputDirNotSetSnafu)?
             .clone();
-        let op = ObjectStore::new(services::Fs::default().root(&root))
+        let op = Operator::new(services::Fs::default().root(&root))
             .context(OpenDalSnafu)?
             .layer(LoggingLayer::default())
             .finish();
@@ -642,14 +642,11 @@ impl Export {
 
     async fn write_to_storage(
         &self,
-        op: &ObjectStore,
+        op: &Operator,
         file_path: &str,
         content: Vec<u8>,
     ) -> Result<()> {
-        op.write(file_path, content)
-            .await
-            .context(OpenDalSnafu)
-            .map(|_| ())
+        op.write(file_path, content).await.context(OpenDalSnafu)
     }
 
     fn get_storage_params(&self, schema: &str) -> (String, String) {

src/cli/src/lib.rs

@@ -17,7 +17,6 @@ mod database;
 pub mod error;
 mod export;
 mod import;
-mod meta_snapshot;
 
 use async_trait::async_trait;
 use clap::Parser;
@@ -28,7 +27,6 @@ use error::Result;
 pub use crate::bench::BenchTableMetadataCommand;
 pub use crate::export::ExportCommand;
 pub use crate::import::ImportCommand;
-pub use crate::meta_snapshot::{MetaRestoreCommand, MetaSnapshotCommand};
 
 #[async_trait]
 pub trait Tool: Send + Sync {

src/cli/src/meta_snapshot.rs

@@ -1,329 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::sync::Arc;
-
-use async_trait::async_trait;
-use clap::Parser;
-use common_base::secrets::{ExposeSecret, SecretString};
-use common_error::ext::BoxedError;
-use common_meta::kv_backend::chroot::ChrootKvBackend;
-use common_meta::kv_backend::etcd::EtcdStore;
-use common_meta::kv_backend::KvBackendRef;
-use common_meta::snapshot::MetadataSnapshotManager;
-use meta_srv::bootstrap::create_etcd_client;
-use meta_srv::metasrv::BackendImpl;
-use object_store::services::{Fs, S3};
-use object_store::ObjectStore;
-use snafu::ResultExt;
-
-use crate::error::{KvBackendNotSetSnafu, OpenDalSnafu, S3ConfigNotSetSnafu};
-use crate::Tool;
-#[derive(Debug, Default, Parser)]
-struct MetaConnection {
-    /// The endpoint of store. one of etcd, pg or mysql.
-    #[clap(long, alias = "store-addr", value_delimiter = ',', num_args = 1..)]
-    store_addrs: Vec<String>,
-    /// The database backend.
-    #[clap(long, value_enum)]
-    backend: Option<BackendImpl>,
-    #[clap(long, default_value = "")]
-    store_key_prefix: String,
-    #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
-    #[clap(long,default_value = common_meta::kv_backend::DEFAULT_META_TABLE_NAME)]
-    meta_table_name: String,
-    #[clap(long, default_value = "128")]
-    max_txn_ops: usize,
-}
-
-impl MetaConnection {
-    pub async fn build(&self) -> Result<KvBackendRef, BoxedError> {
-        let max_txn_ops = self.max_txn_ops;
-        let store_addrs = &self.store_addrs;
-        if store_addrs.is_empty() {
-            KvBackendNotSetSnafu { backend: "all" }
-                .fail()
-                .map_err(BoxedError::new)
-        } else {
-            let kvbackend = match self.backend {
-                Some(BackendImpl::EtcdStore) => {
-                    let etcd_client = create_etcd_client(store_addrs)
-                        .await
-                        .map_err(BoxedError::new)?;
-                    Ok(EtcdStore::with_etcd_client(etcd_client, max_txn_ops))
-                }
-                #[cfg(feature = "pg_kvbackend")]
-                Some(BackendImpl::PostgresStore) => {
-                    let table_name = &self.meta_table_name;
-                    let pool = meta_srv::bootstrap::create_postgres_pool(store_addrs)
-                        .await
-                        .map_err(BoxedError::new)?;
-                    Ok(common_meta::kv_backend::rds::PgStore::with_pg_pool(
-                        pool,
-                        table_name,
-                        max_txn_ops,
-                    )
-                    .await
-                    .map_err(BoxedError::new)?)
-                }
-                #[cfg(feature = "mysql_kvbackend")]
-                Some(BackendImpl::MysqlStore) => {
-                    let table_name = &self.meta_table_name;
-                    let pool = meta_srv::bootstrap::create_mysql_pool(store_addrs)
-                        .await
-                        .map_err(BoxedError::new)?;
-                    Ok(common_meta::kv_backend::rds::MySqlStore::with_mysql_pool(
-                        pool,
-                        table_name,
-                        max_txn_ops,
-                    )
-                    .await
-                    .map_err(BoxedError::new)?)
-                }
-                _ => KvBackendNotSetSnafu { backend: "all" }
-                    .fail()
-                    .map_err(BoxedError::new),
-            };
-            if self.store_key_prefix.is_empty() {
-                kvbackend
-            } else {
-                let chroot_kvbackend =
-                    ChrootKvBackend::new(self.store_key_prefix.as_bytes().to_vec(), kvbackend?);
-                Ok(Arc::new(chroot_kvbackend))
-            }
-        }
-    }
-}
-
-// TODO(qtang): Abstract a generic s3 config for export import meta snapshot restore
-#[derive(Debug, Default, Parser)]
-struct S3Config {
-    /// whether to use s3 as the output directory. default is false.
-    #[clap(long, default_value = "false")]
-    s3: bool,
-    /// The s3 bucket name.
-    #[clap(long)]
-    s3_bucket: Option<String>,
-    /// The s3 region.
-    #[clap(long)]
-    s3_region: Option<String>,
-    /// The s3 access key.
-    #[clap(long)]
-    s3_access_key: Option<SecretString>,
-    /// The s3 secret key.
-    #[clap(long)]
-    s3_secret_key: Option<SecretString>,
-    /// The s3 endpoint. we will automatically use the default s3 decided by the region if not set.
-    #[clap(long)]
-    s3_endpoint: Option<String>,
-}
-
-impl S3Config {
-    pub fn build(&self, root: &str) -> Result<Option<ObjectStore>, BoxedError> {
-        if !self.s3 {
-            Ok(None)
-        } else {
-            if self.s3_region.is_none()
-                || self.s3_access_key.is_none()
-                || self.s3_secret_key.is_none()
-                || self.s3_bucket.is_none()
-            {
-                return S3ConfigNotSetSnafu.fail().map_err(BoxedError::new);
-            }
-            // Safety, unwrap is safe because we have checked the options above.
-            let mut config = S3::default()
-                .bucket(self.s3_bucket.as_ref().unwrap())
-                .region(self.s3_region.as_ref().unwrap())
-                .access_key_id(self.s3_access_key.as_ref().unwrap().expose_secret())
-                .secret_access_key(self.s3_secret_key.as_ref().unwrap().expose_secret());
-
-            if !root.is_empty() && root != "." {
-                config = config.root(root);
-            }
-
-            if let Some(endpoint) = &self.s3_endpoint {
-                config = config.endpoint(endpoint);
-            }
-            Ok(Some(
-                ObjectStore::new(config)
-                    .context(OpenDalSnafu)
-                    .map_err(BoxedError::new)?
-                    .finish(),
-            ))
-        }
-    }
-}
-
-/// Export metadata snapshot tool.
-/// This tool is used to export metadata snapshot from etcd, pg or mysql.
-/// It will dump the metadata snapshot to local file or s3 bucket.
-/// The snapshot file will be in binary format.
-#[derive(Debug, Default, Parser)]
-pub struct MetaSnapshotCommand {
-    /// The connection to the metadata store.
-    #[clap(flatten)]
-    connection: MetaConnection,
-    /// The s3 config.
-    #[clap(flatten)]
-    s3_config: S3Config,
-    /// The name of the target snapshot file. we will add the file extension automatically.
-    #[clap(long, default_value = "metadata_snapshot")]
-    file_name: String,
-    /// The directory to store the snapshot file.
-    /// if target output is s3 bucket, this is the root directory in the bucket.
-    /// if target output is local file, this is the local directory.
-    #[clap(long, default_value = "")]
-    output_dir: String,
-}
-
-fn create_local_file_object_store(root: &str) -> Result<ObjectStore, BoxedError> {
-    let root = if root.is_empty() { "." } else { root };
-    let object_store = ObjectStore::new(Fs::default().root(root))
-        .context(OpenDalSnafu)
-        .map_err(BoxedError::new)?
-        .finish();
-    Ok(object_store)
-}
-
-impl MetaSnapshotCommand {
-    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
-        let kvbackend = self.connection.build().await?;
-        let output_dir = &self.output_dir;
-        let object_store = self.s3_config.build(output_dir).map_err(BoxedError::new)?;
-        if let Some(store) = object_store {
-            let tool = MetaSnapshotTool {
-                inner: MetadataSnapshotManager::new(kvbackend, store),
-                target_file: self.file_name.clone(),
-            };
-            Ok(Box::new(tool))
-        } else {
-            let object_store = create_local_file_object_store(output_dir)?;
-            let tool = MetaSnapshotTool {
-                inner: MetadataSnapshotManager::new(kvbackend, object_store),
-                target_file: self.file_name.clone(),
-            };
-            Ok(Box::new(tool))
-        }
-    }
-}
-
-pub struct MetaSnapshotTool {
-    inner: MetadataSnapshotManager,
-    target_file: String,
-}
-
-#[async_trait]
-impl Tool for MetaSnapshotTool {
-    async fn do_work(&self) -> std::result::Result<(), BoxedError> {
-        self.inner
-            .dump("", &self.target_file)
-            .await
-            .map_err(BoxedError::new)?;
-        Ok(())
-    }
-}
-
-/// Restore metadata snapshot tool.
-/// This tool is used to restore metadata snapshot from etcd, pg or mysql.
-/// It will restore the metadata snapshot from local file or s3 bucket.
-#[derive(Debug, Default, Parser)]
-pub struct MetaRestoreCommand {
-    /// The connection to the metadata store.
-    #[clap(flatten)]
-    connection: MetaConnection,
-    /// The s3 config.
-    #[clap(flatten)]
-    s3_config: S3Config,
-    /// The name of the target snapshot file.
-    #[clap(long, default_value = "metadata_snapshot.metadata.fb")]
-    file_name: String,
-    /// The directory to store the snapshot file.
-    #[clap(long, default_value = ".")]
-    input_dir: String,
-    #[clap(long, default_value = "false")]
-    force: bool,
-}
-
-impl MetaRestoreCommand {
-    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
-        let kvbackend = self.connection.build().await?;
-        let input_dir = &self.input_dir;
-        let object_store = self.s3_config.build(input_dir).map_err(BoxedError::new)?;
-        if let Some(store) = object_store {
-            let tool = MetaRestoreTool::new(
-                MetadataSnapshotManager::new(kvbackend, store),
-                self.file_name.clone(),
-                self.force,
-            );
-            Ok(Box::new(tool))
-        } else {
-            let object_store = create_local_file_object_store(input_dir)?;
-            let tool = MetaRestoreTool::new(
-                MetadataSnapshotManager::new(kvbackend, object_store),
-                self.file_name.clone(),
-                self.force,
-            );
-            Ok(Box::new(tool))
-        }
-    }
-}
-
-pub struct MetaRestoreTool {
-    inner: MetadataSnapshotManager,
-    source_file: String,
-    force: bool,
-}
-
-impl MetaRestoreTool {
-    pub fn new(inner: MetadataSnapshotManager, source_file: String, force: bool) -> Self {
-        Self {
-            inner,
-            source_file,
-            force,
-        }
-    }
-}
-
-#[async_trait]
-impl Tool for MetaRestoreTool {
-    async fn do_work(&self) -> std::result::Result<(), BoxedError> {
-        let clean = self
-            .inner
-            .check_target_source_clean()
-            .await
-            .map_err(BoxedError::new)?;
-        if clean {
-            common_telemetry::info!(
-                "The target source is clean, we will restore the metadata snapshot."
-            );
-            self.inner
-                .restore(&self.source_file)
-                .await
-                .map_err(BoxedError::new)?;
-            Ok(())
-        } else if !self.force {
-            common_telemetry::warn!(
-                 "The target source is not clean, if you want to restore the metadata snapshot forcefully, please use --force option."
-             );
-            Ok(())
-        } else {
-            common_telemetry::info!("The target source is not clean, We will restore the metadata snapshot with --force.");
-            self.inner
-                .restore(&self.source_file)
-                .await
-                .map_err(BoxedError::new)?;
-            Ok(())
-        }
-    }
-}

src/client/Cargo.toml

@@ -25,7 +25,6 @@ common-meta.workspace = true
 common-query.workspace = true
 common-recordbatch.workspace = true
 common-telemetry.workspace = true
-datatypes.workspace = true
 enum_dispatch = "0.3"
 futures.workspace = true
 futures-util.workspace = true

src/client/src/database.rs

@@ -14,7 +14,6 @@
 
 use std::pin::Pin;
 use std::str::FromStr;
-use std::sync::Arc;
 
 use api::v1::auth_header::AuthScheme;
 use api::v1::ddl_request::Expr as DdlExpr;
@@ -36,7 +35,7 @@ use common_grpc::flight::do_put::DoPutResponse;
 use common_grpc::flight::{FlightDecoder, FlightMessage};
 use common_query::Output;
 use common_recordbatch::error::ExternalSnafu;
-use common_recordbatch::{RecordBatch, RecordBatchStreamWrapper};
+use common_recordbatch::RecordBatchStreamWrapper;
 use common_telemetry::tracing_context::W3cTrace;
 use common_telemetry::{error, warn};
 use futures::future;
@@ -50,7 +49,7 @@ use crate::error::{
     ConvertFlightDataSnafu, Error, FlightGetSnafu, IllegalFlightMessagesSnafu,
     InvalidTonicMetadataValueSnafu, ServerSnafu,
 };
-use crate::{error, from_grpc_response, Client, Result};
+use crate::{from_grpc_response, Client, Result};
 
 type FlightDataStream = Pin<Box<dyn Stream<Item = FlightData> + Send>>;
 
@@ -338,30 +337,20 @@ impl Database {
                 );
                 Ok(Output::new_with_affected_rows(rows))
             }
-            FlightMessage::RecordBatch(_) | FlightMessage::Metrics(_) => {
+            FlightMessage::Recordbatch(_) | FlightMessage::Metrics(_) => {
                 IllegalFlightMessagesSnafu {
                     reason: "The first flight message cannot be a RecordBatch or Metrics message",
                 }
                 .fail()
             }
             FlightMessage::Schema(schema) => {
-                let schema = Arc::new(
-                    datatypes::schema::Schema::try_from(schema)
-                        .context(error::ConvertSchemaSnafu)?,
-                );
-                let schema_cloned = schema.clone();
                 let stream = Box::pin(stream!({
                     while let Some(flight_message) = flight_message_stream.next().await {
                         let flight_message = flight_message
                             .map_err(BoxedError::new)
                             .context(ExternalSnafu)?;
                         match flight_message {
-                            FlightMessage::RecordBatch(arrow_batch) => {
-                                yield RecordBatch::try_from_df_record_batch(
-                                    schema_cloned.clone(),
-                                    arrow_batch,
-                                )
-                            }
+                            FlightMessage::Recordbatch(record_batch) => yield Ok(record_batch),
                             FlightMessage::Metrics(_) => {}
                             FlightMessage::AffectedRows(_) | FlightMessage::Schema(_) => {
                                 yield IllegalFlightMessagesSnafu {reason: format!("A Schema message must be succeeded exclusively by a set of RecordBatch messages, flight_message: {:?}", flight_message)}

src/client/src/error.rs

@@ -117,13 +117,6 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
-
-    #[snafu(display("Failed to convert Schema"))]
-    ConvertSchema {
-        #[snafu(implicit)]
-        location: Location,
-        source: datatypes::error::Error,
-    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -144,7 +137,6 @@ impl ErrorExt for Error {
             | Error::CreateTlsChannel { source, .. } => source.status_code(),
             Error::IllegalGrpcClientState { .. } => StatusCode::Unexpected,
             Error::InvalidTonicMetadataValue { .. } => StatusCode::InvalidArguments,
-            Error::ConvertSchema { source, .. } => source.status_code(),
         }
     }
 

src/client/src/region.rs

@@ -28,7 +28,7 @@ use common_meta::error::{self as meta_error, Result as MetaResult};
 use common_meta::node_manager::Datanode;
 use common_query::request::QueryRequest;
 use common_recordbatch::error::ExternalSnafu;
-use common_recordbatch::{RecordBatch, RecordBatchStreamWrapper, SendableRecordBatchStream};
+use common_recordbatch::{RecordBatchStreamWrapper, SendableRecordBatchStream};
 use common_telemetry::error;
 use common_telemetry::tracing_context::TracingContext;
 use prost::Message;
@@ -55,7 +55,6 @@ impl Datanode for RegionRequester {
             if err.should_retry() {
                 meta_error::Error::RetryLater {
                     source: BoxedError::new(err),
-                    clean_poisons: false,
                 }
             } else {
                 meta_error::Error::External {
@@ -147,10 +146,6 @@ impl RegionRequester {
 
         let tracing_context = TracingContext::from_current_span();
 
-        let schema = Arc::new(
-            datatypes::schema::Schema::try_from(schema).context(error::ConvertSchemaSnafu)?,
-        );
-        let schema_cloned = schema.clone();
         let stream = Box::pin(stream!({
             let _span = tracing_context.attach(common_telemetry::tracing::info_span!(
                 "poll_flight_data_stream"
@@ -161,12 +156,7 @@ impl RegionRequester {
                     .context(ExternalSnafu)?;
 
                 match flight_message {
-                    FlightMessage::RecordBatch(record_batch) => {
-                        yield RecordBatch::try_from_df_record_batch(
-                            schema_cloned.clone(),
-                            record_batch,
-                        )
-                    }
+                    FlightMessage::Recordbatch(record_batch) => yield Ok(record_batch),
                     FlightMessage::Metrics(s) => {
                         let m = serde_json::from_str(&s).ok().map(Arc::new);
                         metrics_ref.swap(m);

src/cmd/Cargo.toml

@@ -10,13 +10,7 @@ name = "greptime"
 path = "src/bin/greptime.rs"
 
 [features]
-default = [
-    "servers/pprof",
-    "servers/mem-prof",
-    "meta-srv/pg_kvbackend",
-    "meta-srv/mysql_kvbackend",
-]
-enterprise = ["common-meta/enterprise", "frontend/enterprise", "meta-srv/enterprise"]
+default = ["servers/pprof", "servers/mem-prof"]
 tokio-console = ["common-telemetry/tokio-console"]
 
 [lints]

src/cmd/src/standalone.rs

@@ -35,8 +35,6 @@ 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;
-#[cfg(feature = "enterprise")]
-use common_meta::ddl_manager::TriggerDdlManagerRef;
 use common_meta::key::flow::flow_state::FlowStat;
 use common_meta::key::flow::{FlowMetadataManager, FlowMetadataManagerRef};
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
@@ -71,7 +69,6 @@ use frontend::service_config::{
 };
 use meta_srv::metasrv::{FLOW_ID_SEQ, TABLE_ID_SEQ};
 use mito2::config::MitoConfig;
-use query::options::QueryOptions;
 use serde::{Deserialize, Serialize};
 use servers::export_metrics::{ExportMetricsOption, ExportMetricsTask};
 use servers::grpc::GrpcOptions;
@@ -156,7 +153,6 @@ pub struct StandaloneOptions {
     pub init_regions_parallelism: usize,
     pub max_in_flight_write_bytes: Option<ReadableSize>,
     pub slow_query: Option<SlowQueryOptions>,
-    pub query: QueryOptions,
 }
 
 impl Default for StandaloneOptions {
@@ -189,7 +185,6 @@ impl Default for StandaloneOptions {
             init_regions_parallelism: 16,
             max_in_flight_write_bytes: None,
             slow_query: Some(SlowQueryOptions::default()),
-            query: QueryOptions::default(),
         }
     }
 }
@@ -245,7 +240,6 @@ impl StandaloneOptions {
             grpc: cloned_opts.grpc,
             init_regions_in_background: cloned_opts.init_regions_in_background,
             init_regions_parallelism: cloned_opts.init_regions_parallelism,
-            query: cloned_opts.query,
             ..Default::default()
         }
     }
@@ -585,8 +579,6 @@ impl StartCommand {
             flow_id_sequence,
         ));
 
-        #[cfg(feature = "enterprise")]
-        let trigger_ddl_manager: Option<TriggerDdlManagerRef> = plugins.get();
         let ddl_task_executor = Self::create_ddl_task_executor(
             procedure_manager.clone(),
             node_manager.clone(),
@@ -595,8 +587,6 @@ impl StartCommand {
             table_meta_allocator,
             flow_metadata_manager,
             flow_meta_allocator,
-            #[cfg(feature = "enterprise")]
-            trigger_ddl_manager,
         )
         .await?;
 
@@ -661,7 +651,6 @@ impl StartCommand {
         })
     }
 
-    #[allow(clippy::too_many_arguments)]
     pub async fn create_ddl_task_executor(
         procedure_manager: ProcedureManagerRef,
         node_manager: NodeManagerRef,
@@ -670,7 +659,6 @@ impl StartCommand {
         table_metadata_allocator: TableMetadataAllocatorRef,
         flow_metadata_manager: FlowMetadataManagerRef,
         flow_metadata_allocator: FlowMetadataAllocatorRef,
-        #[cfg(feature = "enterprise")] trigger_ddl_manager: Option<TriggerDdlManagerRef>,
     ) -> Result<ProcedureExecutorRef> {
         let procedure_executor: ProcedureExecutorRef = Arc::new(
             DdlManager::try_new(
@@ -687,8 +675,6 @@ impl StartCommand {
                 },
                 procedure_manager,
                 true,
-                #[cfg(feature = "enterprise")]
-                trigger_ddl_manager,
             )
             .context(error::InitDdlManagerSnafu)?,
         );

src/common/function/src/adjust_flow.rs

@@ -1,90 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use common_macro::admin_fn;
-use common_query::error::{
-    InvalidFuncArgsSnafu, MissingFlowServiceHandlerSnafu, Result, UnsupportedInputDataTypeSnafu,
-};
-use common_query::prelude::Signature;
-use datafusion::logical_expr::Volatility;
-use datatypes::value::{Value, ValueRef};
-use session::context::QueryContextRef;
-use snafu::ensure;
-use store_api::storage::ConcreteDataType;
-
-use crate::handlers::FlowServiceHandlerRef;
-use crate::helper::parse_catalog_flow;
-
-fn adjust_signature() -> Signature {
-    Signature::exact(
-        vec![
-            ConcreteDataType::string_datatype(), // flow name
-            ConcreteDataType::uint64_datatype(), // min_run_interval in seconds
-            ConcreteDataType::uint64_datatype(), // max filter number per query
-        ],
-        Volatility::Immutable,
-    )
-}
-
-#[admin_fn(
-    name = AdjustFlowFunction,
-    display_name = adjust_flow,
-    sig_fn = adjust_signature,
-    ret = uint64
-)]
-pub(crate) async fn adjust_flow(
-    flow_service_handler: &FlowServiceHandlerRef,
-    query_ctx: &QueryContextRef,
-    params: &[ValueRef<'_>],
-) -> Result<Value> {
-    ensure!(
-        params.len() == 3,
-        InvalidFuncArgsSnafu {
-            err_msg: format!(
-                "The length of the args is not correct, expect 3, have: {}",
-                params.len()
-            ),
-        }
-    );
-
-    let (flow_name, min_run_interval, max_filter_num) = match (params[0], params[1], params[2]) {
-        (
-            ValueRef::String(flow_name),
-            ValueRef::UInt64(min_run_interval),
-            ValueRef::UInt64(max_filter_num),
-        ) => (flow_name, min_run_interval, max_filter_num),
-        _ => {
-            return UnsupportedInputDataTypeSnafu {
-                function: "adjust_flow",
-                datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
-            }
-            .fail();
-        }
-    };
-
-    let (catalog_name, flow_name) = parse_catalog_flow(flow_name, query_ctx)?;
-
-    let res = flow_service_handler
-        .adjust(
-            &catalog_name,
-            &flow_name,
-            min_run_interval,
-            max_filter_num as usize,
-            query_ctx.clone(),
-        )
-        .await?;
-    let affected_rows = res.affected_rows;
-
-    Ok(Value::from(affected_rows))
-}

src/common/function/src/admin.rs

@@ -26,7 +26,6 @@ use flush_compact_table::{CompactTableFunction, FlushTableFunction};
 use migrate_region::MigrateRegionFunction;
 use remove_region_follower::RemoveRegionFollowerFunction;
 
-use crate::adjust_flow::AdjustFlowFunction;
 use crate::flush_flow::FlushFlowFunction;
 use crate::function_registry::FunctionRegistry;
 
@@ -44,6 +43,5 @@ impl AdminFunction {
         registry.register_async(Arc::new(FlushTableFunction));
         registry.register_async(Arc::new(CompactTableFunction));
         registry.register_async(Arc::new(FlushFlowFunction));
-        registry.register_async(Arc::new(AdjustFlowFunction));
     }
 }

src/common/function/src/flush_flow.rs

@@ -12,19 +12,21 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use common_error::ext::BoxedError;
 use common_macro::admin_fn;
 use common_query::error::{
-    InvalidFuncArgsSnafu, MissingFlowServiceHandlerSnafu, Result, UnsupportedInputDataTypeSnafu,
+    ExecuteSnafu, InvalidFuncArgsSnafu, MissingFlowServiceHandlerSnafu, Result,
+    UnsupportedInputDataTypeSnafu,
 };
 use common_query::prelude::Signature;
 use datafusion::logical_expr::Volatility;
 use datatypes::value::{Value, ValueRef};
 use session::context::QueryContextRef;
-use snafu::ensure;
+use snafu::{ensure, ResultExt};
+use sql::parser::ParserContext;
 use store_api::storage::ConcreteDataType;
 
 use crate::handlers::FlowServiceHandlerRef;
-use crate::helper::parse_catalog_flow;
 
 fn flush_signature() -> Signature {
     Signature::uniform(
@@ -45,6 +47,20 @@ pub(crate) async fn flush_flow(
     query_ctx: &QueryContextRef,
     params: &[ValueRef<'_>],
 ) -> Result<Value> {
+    let (catalog_name, flow_name) = parse_flush_flow(params, query_ctx)?;
+
+    let res = flow_service_handler
+        .flush(&catalog_name, &flow_name, query_ctx.clone())
+        .await?;
+    let affected_rows = res.affected_rows;
+
+    Ok(Value::from(affected_rows))
+}
+
+fn parse_flush_flow(
+    params: &[ValueRef<'_>],
+    query_ctx: &QueryContextRef,
+) -> Result<(String, String)> {
     ensure!(
         params.len() == 1,
         InvalidFuncArgsSnafu {
@@ -54,6 +70,7 @@ pub(crate) async fn flush_flow(
             ),
         }
     );
+
     let ValueRef::String(flow_name) = params[0] else {
         return UnsupportedInputDataTypeSnafu {
             function: "flush_flow",
@@ -61,14 +78,27 @@ pub(crate) async fn flush_flow(
         }
         .fail();
     };
-    let (catalog_name, flow_name) = parse_catalog_flow(flow_name, query_ctx)?;
+    let obj_name = ParserContext::parse_table_name(flow_name, query_ctx.sql_dialect())
+        .map_err(BoxedError::new)
+        .context(ExecuteSnafu)?;
 
-    let res = flow_service_handler
-        .flush(&catalog_name, &flow_name, query_ctx.clone())
-        .await?;
-    let affected_rows = res.affected_rows;
-
-    Ok(Value::from(affected_rows))
+    let (catalog_name, flow_name) = match &obj_name.0[..] {
+        [flow_name] => (
+            query_ctx.current_catalog().to_string(),
+            flow_name.value.clone(),
+        ),
+        [catalog, flow_name] => (catalog.value.clone(), flow_name.value.clone()),
+        _ => {
+            return InvalidFuncArgsSnafu {
+                err_msg: format!(
+                    "expect flow name to be <catalog>.<flow-name> or <flow-name>, actual: {}",
+                    obj_name
+                ),
+            }
+            .fail()
+        }
+    };
+    Ok((catalog_name, flow_name))
 }
 
 #[cfg(test)]
@@ -124,7 +154,10 @@ mod test {
             ("catalog.flow_name", ("catalog", "flow_name")),
         ];
         for (input, expected) in testcases.iter() {
-            let result = parse_catalog_flow(input, &QueryContext::arc()).unwrap();
+            let args = vec![*input];
+            let args = args.into_iter().map(ValueRef::String).collect::<Vec<_>>();
+
+            let result = parse_flush_flow(&args, &QueryContext::arc()).unwrap();
             assert_eq!(*expected, (result.0.as_str(), result.1.as_str()));
         }
     }

src/common/function/src/handlers.rs

@@ -87,15 +87,6 @@ pub trait FlowServiceHandler: Send + Sync {
         flow: &str,
         ctx: QueryContextRef,
     ) -> Result<api::v1::flow::FlowResponse>;
-
-    async fn adjust(
-        &self,
-        catalog: &str,
-        flow: &str,
-        min_run_interval_secs: u64,
-        max_filter_num_per_query: usize,
-        ctx: QueryContextRef,
-    ) -> Result<api::v1::flow::FlowResponse>;
 }
 
 pub type TableMutationHandlerRef = Arc<dyn TableMutationHandler>;

src/common/function/src/helper.rs

@@ -12,15 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use common_error::ext::BoxedError;
-use common_query::error::{ExecuteSnafu, InvalidFuncArgsSnafu, InvalidInputTypeSnafu, Result};
+use common_query::error::{InvalidInputTypeSnafu, Result};
 use common_query::prelude::{Signature, TypeSignature, Volatility};
 use datatypes::prelude::ConcreteDataType;
 use datatypes::types::cast::cast;
 use datatypes::value::ValueRef;
-use session::context::QueryContextRef;
 use snafu::ResultExt;
-use sql::parser::ParserContext;
 
 /// Create a function signature with oneof signatures of interleaving two arguments.
 pub fn one_of_sigs2(args1: Vec<ConcreteDataType>, args2: Vec<ConcreteDataType>) -> Signature {
@@ -46,30 +43,3 @@ pub fn cast_u64(value: &ValueRef) -> Result<Option<u64>> {
         })
         .map(|v| v.as_u64())
 }
-
-pub fn parse_catalog_flow(
-    flow_name: &str,
-    query_ctx: &QueryContextRef,
-) -> Result<(String, String)> {
-    let obj_name = ParserContext::parse_table_name(flow_name, query_ctx.sql_dialect())
-        .map_err(BoxedError::new)
-        .context(ExecuteSnafu)?;
-
-    let (catalog_name, flow_name) = match &obj_name.0[..] {
-        [flow_name] => (
-            query_ctx.current_catalog().to_string(),
-            flow_name.value.clone(),
-        ),
-        [catalog, flow_name] => (catalog.value.clone(), flow_name.value.clone()),
-        _ => {
-            return InvalidFuncArgsSnafu {
-                err_msg: format!(
-                    "expect flow name to be <catalog>.<flow-name> or <flow-name>, actual: {}",
-                    obj_name
-                ),
-            }
-            .fail()
-        }
-    };
-    Ok((catalog_name, flow_name))
-}

src/common/function/src/lib.rs

@@ -15,7 +15,6 @@
 #![feature(let_chains)]
 #![feature(try_blocks)]
 
-mod adjust_flow;
 mod admin;
 mod flush_flow;
 mod macros;

src/common/function/src/state.rs

@@ -148,17 +148,6 @@ impl FunctionState {
             ) -> Result<api::v1::flow::FlowResponse> {
                 todo!()
             }
-
-            async fn adjust(
-                &self,
-                _catalog: &str,
-                _flow: &str,
-                _min_run_interval_secs: u64,
-                _max_filter_num_per_query: usize,
-                _ctx: QueryContextRef,
-            ) -> Result<api::v1::flow::FlowResponse> {
-                todo!()
-            }
         }
 
         Self {

src/common/grpc/Cargo.toml

@@ -10,7 +10,6 @@ workspace = true
 [dependencies]
 api.workspace = true
 arrow-flight.workspace = true
-bytes.workspace = true
 common-base.workspace = true
 common-error.workspace = true
 common-macro.workspace = true

src/common/grpc/benches/bench_flight_decoder.rs

@@ -1,146 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::sync::Arc;
-
-use arrow_flight::FlightData;
-use bytes::Bytes;
-use common_grpc::flight::{FlightDecoder, FlightEncoder, FlightMessage};
-use common_recordbatch::DfRecordBatch;
-use criterion::{criterion_group, criterion_main, Criterion};
-use datatypes::arrow;
-use datatypes::arrow::array::{ArrayRef, Int64Array, StringArray, TimestampMillisecondArray};
-use datatypes::arrow::datatypes::DataType;
-use datatypes::data_type::ConcreteDataType;
-use datatypes::schema::{ColumnSchema, Schema};
-use prost::Message;
-
-fn schema() -> arrow::datatypes::SchemaRef {
-    let schema = Schema::new(vec![
-        ColumnSchema::new("k0", ConcreteDataType::string_datatype(), false),
-        ColumnSchema::new("k1", ConcreteDataType::string_datatype(), false),
-        ColumnSchema::new(
-            "ts",
-            ConcreteDataType::timestamp_millisecond_datatype(),
-            false,
-        ),
-        ColumnSchema::new("v0", ConcreteDataType::int64_datatype(), false),
-        ColumnSchema::new("v1", ConcreteDataType::int64_datatype(), false),
-    ]);
-    schema.arrow_schema().clone()
-}
-
-/// Generate record batch according to provided schema and num rows.
-fn prepare_random_record_batch(
-    schema: arrow::datatypes::SchemaRef,
-    num_rows: usize,
-) -> DfRecordBatch {
-    let tag_candidates = (0..10000).map(|i| i.to_string()).collect::<Vec<_>>();
-
-    let columns: Vec<ArrayRef> = schema
-        .fields
-        .iter()
-        .map(|col| match col.data_type() {
-            DataType::Utf8 => {
-                let array = StringArray::from(
-                    (0..num_rows)
-                        .map(|_| {
-                            let idx: usize = rand::random_range(0..10000);
-                            format!("tag-{}", tag_candidates[idx])
-                        })
-                        .collect::<Vec<_>>(),
-                );
-                Arc::new(array) as ArrayRef
-            }
-            DataType::Timestamp(_, _) => {
-                let now = common_time::util::current_time_millis();
-                let array = TimestampMillisecondArray::from(
-                    (0..num_rows).map(|i| now + i as i64).collect::<Vec<_>>(),
-                );
-                Arc::new(array) as ArrayRef
-            }
-            DataType::Int64 => {
-                let array = Int64Array::from((0..num_rows).map(|i| i as i64).collect::<Vec<_>>());
-                Arc::new(array) as ArrayRef
-            }
-            _ => unreachable!(),
-        })
-        .collect();
-
-    DfRecordBatch::try_new(schema, columns).unwrap()
-}
-
-fn prepare_flight_data(num_rows: usize) -> (FlightData, FlightData) {
-    let schema = schema();
-    let mut encoder = FlightEncoder::default();
-    let schema_data = encoder.encode(FlightMessage::Schema(schema.clone()));
-    let rb = prepare_random_record_batch(schema, num_rows);
-    let rb_data = encoder.encode(FlightMessage::RecordBatch(rb));
-    (schema_data, rb_data)
-}
-
-fn decode_flight_data_from_protobuf(schema: &Bytes, payload: &Bytes) -> DfRecordBatch {
-    let schema = FlightData::decode(&schema[..]).unwrap();
-    let payload = FlightData::decode(&payload[..]).unwrap();
-    let mut decoder = FlightDecoder::default();
-    let _schema = decoder.try_decode(&schema).unwrap();
-    let message = decoder.try_decode(&payload).unwrap();
-    let FlightMessage::RecordBatch(batch) = message else {
-        unreachable!("unexpected message");
-    };
-    batch
-}
-
-fn decode_flight_data_from_header_and_body(
-    schema: &Bytes,
-    data_header: &Bytes,
-    data_body: &Bytes,
-) -> DfRecordBatch {
-    let mut decoder = FlightDecoder::try_from_schema_bytes(schema).unwrap();
-    decoder
-        .try_decode_record_batch(data_header, data_body)
-        .unwrap()
-}
-
-fn bench_decode_flight_data(c: &mut Criterion) {
-    let row_counts = [100000, 200000, 1000000];
-
-    for row_count in row_counts {
-        let (schema, payload) = prepare_flight_data(row_count);
-
-        // arguments for decode_flight_data_from_protobuf
-        let schema_bytes = Bytes::from(schema.encode_to_vec());
-        let payload_bytes = Bytes::from(payload.encode_to_vec());
-
-        let mut group = c.benchmark_group(format!("flight_decoder_{}_rows", row_count));
-        group.bench_function("decode_from_protobuf", |b| {
-            b.iter(|| decode_flight_data_from_protobuf(&schema_bytes, &payload_bytes));
-        });
-
-        group.bench_function("decode_from_header_and_body", |b| {
-            b.iter(|| {
-                decode_flight_data_from_header_and_body(
-                    &schema.data_header,
-                    &payload.data_header,
-                    &payload.data_body,
-                )
-            });
-        });
-
-        group.finish();
-    }
-}
-
-criterion_group!(benches, bench_decode_flight_data);
-criterion_main!(benches);

src/common/grpc/benches/bench_main.rs

@@ -14,10 +14,8 @@
 
 use criterion::criterion_main;
 
-mod bench_flight_decoder;
 mod channel_manager;
 
 criterion_main! {
-    channel_manager::benches,
-    bench_flight_decoder::benches
+    channel_manager::benches
 }

src/common/grpc/src/error.rs

@@ -18,7 +18,6 @@ use std::io;
 use common_error::ext::ErrorExt;
 use common_error::status_code::StatusCode;
 use common_macro::stack_trace_debug;
-use datatypes::arrow::error::ArrowError;
 use snafu::{Location, Snafu};
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -60,6 +59,13 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Failed to create RecordBatch"))]
+    CreateRecordBatch {
+        #[snafu(implicit)]
+        location: Location,
+        source: common_recordbatch::error::Error,
+    },
+
     #[snafu(display("Failed to convert Arrow type: {}", from))]
     Conversion {
         from: String,
@@ -82,6 +88,13 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Failed to convert Arrow Schema"))]
+    ConvertArrowSchema {
+        #[snafu(implicit)]
+        location: Location,
+        source: datatypes::error::Error,
+    },
+
     #[snafu(display("Not supported: {}", feat))]
     NotSupported { feat: String },
 
@@ -92,14 +105,6 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
-
-    #[snafu(display("Failed arrow operation"))]
-    Arrow {
-        #[snafu(implicit)]
-        location: Location,
-        #[snafu(source)]
-        error: ArrowError,
-    },
 }
 
 impl ErrorExt for Error {
@@ -116,7 +121,8 @@ impl ErrorExt for Error {
             | Error::DecodeFlightData { .. }
             | Error::SerdeJson { .. } => StatusCode::Internal,
 
-            Error::Arrow { .. } => StatusCode::Internal,
+            Error::CreateRecordBatch { source, .. } => source.status_code(),
+            Error::ConvertArrowSchema { source, .. } => source.status_code(),
         }
     }
 

src/common/grpc/src/flight.rs

@@ -21,24 +21,25 @@ use api::v1::{AffectedRows, FlightMetadata, Metrics};
 use arrow_flight::utils::flight_data_to_arrow_batch;
 use arrow_flight::{FlightData, SchemaAsIpc};
 use common_base::bytes::Bytes;
-use common_recordbatch::DfRecordBatch;
+use common_recordbatch::{RecordBatch, RecordBatches};
 use datatypes::arrow;
-use datatypes::arrow::buffer::Buffer;
-use datatypes::arrow::datatypes::{Schema as ArrowSchema, SchemaRef};
-use datatypes::arrow::error::ArrowError;
-use datatypes::arrow::ipc::{convert, reader, root_as_message, writer, MessageHeader};
+use datatypes::arrow::datatypes::Schema as ArrowSchema;
+use datatypes::arrow::ipc::{root_as_message, writer, MessageHeader};
+use datatypes::schema::{Schema, SchemaRef};
 use flatbuffers::FlatBufferBuilder;
 use prost::bytes::Bytes as ProstBytes;
 use prost::Message;
 use snafu::{OptionExt, ResultExt};
 
-use crate::error;
-use crate::error::{DecodeFlightDataSnafu, InvalidFlightDataSnafu, Result};
+use crate::error::{
+    ConvertArrowSchemaSnafu, CreateRecordBatchSnafu, DecodeFlightDataSnafu, InvalidFlightDataSnafu,
+    Result,
+};
 
 #[derive(Debug, Clone)]
 pub enum FlightMessage {
     Schema(SchemaRef),
-    RecordBatch(DfRecordBatch),
+    Recordbatch(RecordBatch),
     AffectedRows(usize),
     Metrics(String),
 }
@@ -66,12 +67,14 @@ impl Default for FlightEncoder {
 impl FlightEncoder {
     pub fn encode(&mut self, flight_message: FlightMessage) -> FlightData {
         match flight_message {
-            FlightMessage::Schema(schema) => SchemaAsIpc::new(&schema, &self.write_options).into(),
-            FlightMessage::RecordBatch(record_batch) => {
+            FlightMessage::Schema(schema) => {
+                SchemaAsIpc::new(schema.arrow_schema(), &self.write_options).into()
+            }
+            FlightMessage::Recordbatch(recordbatch) => {
                 let (encoded_dictionaries, encoded_batch) = self
                     .data_gen
                     .encoded_batch(
-                        &record_batch,
+                        recordbatch.df_record_batch(),
                         &mut self.dictionary_tracker,
                         &self.write_options,
                     )
@@ -121,58 +124,9 @@ impl FlightEncoder {
 #[derive(Default)]
 pub struct FlightDecoder {
     schema: Option<SchemaRef>,
-    schema_bytes: Option<bytes::Bytes>,
 }
 
 impl FlightDecoder {
-    /// Build a [FlightDecoder] instance from provided schema bytes.
-    pub fn try_from_schema_bytes(schema_bytes: &bytes::Bytes) -> Result<Self> {
-        let arrow_schema = convert::try_schema_from_flatbuffer_bytes(&schema_bytes[..])
-            .context(error::ArrowSnafu)?;
-        Ok(Self {
-            schema: Some(Arc::new(arrow_schema)),
-            schema_bytes: Some(schema_bytes.clone()),
-        })
-    }
-
-    pub fn try_decode_record_batch(
-        &mut self,
-        data_header: &bytes::Bytes,
-        data_body: &bytes::Bytes,
-    ) -> Result<DfRecordBatch> {
-        let schema = self
-            .schema
-            .as_ref()
-            .context(InvalidFlightDataSnafu {
-                reason: "Should have decoded schema first!",
-            })?
-            .clone();
-        let message = root_as_message(&data_header[..])
-            .map_err(|err| {
-                ArrowError::ParseError(format!("Unable to get root as message: {err:?}"))
-            })
-            .context(error::ArrowSnafu)?;
-        let result = message
-            .header_as_record_batch()
-            .ok_or_else(|| {
-                ArrowError::ParseError(
-                    "Unable to convert flight data header to a record batch".to_string(),
-                )
-            })
-            .and_then(|batch| {
-                reader::read_record_batch(
-                    &Buffer::from(data_body.as_ref()),
-                    batch,
-                    schema,
-                    &HashMap::new(),
-                    None,
-                    &message.version(),
-                )
-            })
-            .context(error::ArrowSnafu)?;
-        Ok(result)
-    }
-
     pub fn try_decode(&mut self, flight_data: &FlightData) -> Result<FlightMessage> {
         let message = root_as_message(&flight_data.data_header).map_err(|e| {
             InvalidFlightDataSnafu {
@@ -198,29 +152,36 @@ impl FlightDecoder {
                 .fail()
             }
             MessageHeader::Schema => {
-                let arrow_schema = Arc::new(ArrowSchema::try_from(flight_data).map_err(|e| {
+                let arrow_schema = ArrowSchema::try_from(flight_data).map_err(|e| {
                     InvalidFlightDataSnafu {
                         reason: e.to_string(),
                     }
                     .build()
-                })?);
-                self.schema = Some(arrow_schema.clone());
-                self.schema_bytes = Some(flight_data.data_header.clone());
-                Ok(FlightMessage::Schema(arrow_schema))
+                })?;
+                let schema =
+                    Arc::new(Schema::try_from(arrow_schema).context(ConvertArrowSchemaSnafu)?);
+
+                self.schema = Some(schema.clone());
+
+                Ok(FlightMessage::Schema(schema))
             }
             MessageHeader::RecordBatch => {
                 let schema = self.schema.clone().context(InvalidFlightDataSnafu {
                     reason: "Should have decoded schema first!",
                 })?;
+                let arrow_schema = schema.arrow_schema().clone();
+
                 let arrow_batch =
-                    flight_data_to_arrow_batch(flight_data, schema.clone(), &HashMap::new())
+                    flight_data_to_arrow_batch(flight_data, arrow_schema, &HashMap::new())
                         .map_err(|e| {
                             InvalidFlightDataSnafu {
                                 reason: e.to_string(),
                             }
                             .build()
                         })?;
-                Ok(FlightMessage::RecordBatch(arrow_batch))
+                let recordbatch = RecordBatch::try_from_df_record_batch(schema, arrow_batch)
+                    .context(CreateRecordBatchSnafu)?;
+                Ok(FlightMessage::Recordbatch(recordbatch))
             }
             other => {
                 let name = other.variant_name().unwrap_or("UNKNOWN");
@@ -235,22 +196,16 @@ impl FlightDecoder {
     pub fn schema(&self) -> Option<&SchemaRef> {
         self.schema.as_ref()
     }
-
-    pub fn schema_bytes(&self) -> Option<bytes::Bytes> {
-        self.schema_bytes.clone()
-    }
 }
 
-pub fn flight_messages_to_recordbatches(
-    messages: Vec<FlightMessage>,
-) -> Result<Vec<DfRecordBatch>> {
+pub fn flight_messages_to_recordbatches(messages: Vec<FlightMessage>) -> Result<RecordBatches> {
     if messages.is_empty() {
-        Ok(vec![])
+        Ok(RecordBatches::empty())
     } else {
         let mut recordbatches = Vec::with_capacity(messages.len() - 1);
 
-        match &messages[0] {
-            FlightMessage::Schema(_schema) => {}
+        let schema = match &messages[0] {
+            FlightMessage::Schema(schema) => schema.clone(),
             _ => {
                 return InvalidFlightDataSnafu {
                     reason: "First Flight Message must be schema!",
@@ -261,7 +216,7 @@ pub fn flight_messages_to_recordbatches(
 
         for message in messages.into_iter().skip(1) {
             match message {
-                FlightMessage::RecordBatch(recordbatch) => recordbatches.push(recordbatch),
+                FlightMessage::Recordbatch(recordbatch) => recordbatches.push(recordbatch),
                 _ => {
                     return InvalidFlightDataSnafu {
                         reason: "Expect the following Flight Messages are all Recordbatches!",
@@ -271,7 +226,7 @@ pub fn flight_messages_to_recordbatches(
             }
         }
 
-        Ok(recordbatches)
+        RecordBatches::try_new(schema, recordbatches).context(CreateRecordBatchSnafu)
     }
 }
 
@@ -292,33 +247,38 @@ fn build_none_flight_msg() -> Bytes {
 #[cfg(test)]
 mod test {
     use arrow_flight::utils::batches_to_flight_data;
-    use datatypes::arrow::array::Int32Array;
-    use datatypes::arrow::datatypes::{DataType, Field, Schema};
+    use datatypes::arrow::datatypes::{DataType, Field};
+    use datatypes::prelude::ConcreteDataType;
+    use datatypes::schema::ColumnSchema;
+    use datatypes::vectors::Int32Vector;
 
     use super::*;
     use crate::Error;
 
     #[test]
     fn test_try_decode() {
-        let schema = Arc::new(ArrowSchema::new(vec![Field::new(
-            "n",
-            DataType::Int32,
-            true,
-        )]));
+        let arrow_schema = ArrowSchema::new(vec![Field::new("n", DataType::Int32, true)]);
+        let schema = Arc::new(Schema::try_from(arrow_schema.clone()).unwrap());
 
-        let batch1 = DfRecordBatch::try_new(
+        let batch1 = RecordBatch::new(
             schema.clone(),
-            vec![Arc::new(Int32Array::from(vec![Some(1), None, Some(3)])) as _],
+            vec![Arc::new(Int32Vector::from(vec![Some(1), None, Some(3)])) as _],
         )
         .unwrap();
-        let batch2 = DfRecordBatch::try_new(
+        let batch2 = RecordBatch::new(
             schema.clone(),
-            vec![Arc::new(Int32Array::from(vec![None, Some(5)])) as _],
+            vec![Arc::new(Int32Vector::from(vec![None, Some(5)])) as _],
         )
         .unwrap();
 
-        let flight_data =
-            batches_to_flight_data(&schema, vec![batch1.clone(), batch2.clone()]).unwrap();
+        let flight_data = batches_to_flight_data(
+            &arrow_schema,
+            vec![
+                batch1.clone().into_df_record_batch(),
+                batch2.clone().into_df_record_batch(),
+            ],
+        )
+        .unwrap();
         assert_eq!(flight_data.len(), 3);
         let [d1, d2, d3] = flight_data.as_slice() else {
             unreachable!()
@@ -344,15 +304,15 @@ mod test {
         let _ = decoder.schema.as_ref().unwrap();
 
         let message = decoder.try_decode(d2).unwrap();
-        assert!(matches!(message, FlightMessage::RecordBatch(_)));
-        let FlightMessage::RecordBatch(actual_batch) = message else {
+        assert!(matches!(message, FlightMessage::Recordbatch(_)));
+        let FlightMessage::Recordbatch(actual_batch) = message else {
             unreachable!()
         };
         assert_eq!(actual_batch, batch1);
 
         let message = decoder.try_decode(d3).unwrap();
-        assert!(matches!(message, FlightMessage::RecordBatch(_)));
-        let FlightMessage::RecordBatch(actual_batch) = message else {
+        assert!(matches!(message, FlightMessage::Recordbatch(_)));
+        let FlightMessage::Recordbatch(actual_batch) = message else {
             unreachable!()
         };
         assert_eq!(actual_batch, batch2);
@@ -360,22 +320,27 @@ mod test {
 
     #[test]
     fn test_flight_messages_to_recordbatches() {
-        let schema = Arc::new(Schema::new(vec![Field::new("m", DataType::Int32, true)]));
-        let batch1 = DfRecordBatch::try_new(
+        let schema = Arc::new(Schema::new(vec![ColumnSchema::new(
+            "m",
+            ConcreteDataType::int32_datatype(),
+            true,
+        )]));
+        let batch1 = RecordBatch::new(
             schema.clone(),
-            vec![Arc::new(Int32Array::from(vec![Some(2), None, Some(4)])) as _],
+            vec![Arc::new(Int32Vector::from(vec![Some(2), None, Some(4)])) as _],
         )
         .unwrap();
-        let batch2 = DfRecordBatch::try_new(
+        let batch2 = RecordBatch::new(
             schema.clone(),
-            vec![Arc::new(Int32Array::from(vec![None, Some(6)])) as _],
+            vec![Arc::new(Int32Vector::from(vec![None, Some(6)])) as _],
         )
         .unwrap();
-        let recordbatches = vec![batch1.clone(), batch2.clone()];
+        let recordbatches =
+            RecordBatches::try_new(schema.clone(), vec![batch1.clone(), batch2.clone()]).unwrap();
 
         let m1 = FlightMessage::Schema(schema);
-        let m2 = FlightMessage::RecordBatch(batch1);
-        let m3 = FlightMessage::RecordBatch(batch2);
+        let m2 = FlightMessage::Recordbatch(batch1);
+        let m3 = FlightMessage::Recordbatch(batch2);
 
         let result = flight_messages_to_recordbatches(vec![m2.clone(), m1.clone(), m3.clone()]);
         assert!(matches!(result, Err(Error::InvalidFlightData { .. })));

src/common/meta/Cargo.toml

@@ -8,7 +8,6 @@ license.workspace = true
 testing = []
 pg_kvbackend = ["dep:tokio-postgres", "dep:backon", "dep:deadpool-postgres", "dep:deadpool"]
 mysql_kvbackend = ["dep:sqlx", "dep:backon"]
-enterprise = []
 
 [lints]
 workspace = true
@@ -43,7 +42,6 @@ deadpool = { workspace = true, optional = true }
 deadpool-postgres = { workspace = true, optional = true }
 derive_builder.workspace = true
 etcd-client.workspace = true
-flexbuffers = "25.2"
 futures.workspace = true
 futures-util.workspace = true
 hex.workspace = true
@@ -51,7 +49,6 @@ humantime-serde.workspace = true
 itertools.workspace = true
 lazy_static.workspace = true
 moka.workspace = true
-object-store.workspace = true
 prometheus.workspace = true
 prost.workspace = true
 rand.workspace = true
@@ -74,7 +71,6 @@ typetag.workspace = true
 [dev-dependencies]
 chrono.workspace = true
 common-procedure = { workspace = true, features = ["testing"] }
-common-test-util.workspace = true
 common-wal = { workspace = true, features = ["testing"] }
 datatypes.workspace = true
 hyper = { version = "0.14", features = ["full"] }

src/common/meta/src/cache_invalidator.rs

@@ -16,12 +16,9 @@ use std::sync::Arc;
 
 use crate::error::Result;
 use crate::flow_name::FlowName;
-use crate::instruction::{CacheIdent, DropFlow};
+use crate::instruction::CacheIdent;
 use crate::key::flow::flow_info::FlowInfoKey;
 use crate::key::flow::flow_name::FlowNameKey;
-use crate::key::flow::flow_route::FlowRouteKey;
-use crate::key::flow::flownode_flow::FlownodeFlowKey;
-use crate::key::flow::table_flow::TableFlowKey;
 use crate::key::schema_name::SchemaNameKey;
 use crate::key::table_info::TableInfoKey;
 use crate::key::table_name::TableNameKey;
@@ -92,40 +89,9 @@ where
                     let key: SchemaNameKey = schema_name.into();
                     self.invalidate_key(&key.to_bytes()).await;
                 }
-                CacheIdent::CreateFlow(_) => {
+                CacheIdent::CreateFlow(_) | CacheIdent::DropFlow(_) => {
                     // Do nothing
                 }
-                CacheIdent::DropFlow(DropFlow {
-                    flow_id,
-                    source_table_ids,
-                    flow_part2node_id,
-                }) => {
-                    // invalidate flow route/flownode flow/table flow
-                    let mut keys = Vec::with_capacity(
-                        source_table_ids.len() * flow_part2node_id.len()
-                            + flow_part2node_id.len() * 2,
-                    );
-                    for table_id in source_table_ids {
-                        for (partition_id, node_id) in flow_part2node_id {
-                            let key =
-                                TableFlowKey::new(*table_id, *node_id, *flow_id, *partition_id)
-                                    .to_bytes();
-                            keys.push(key);
-                        }
-                    }
-
-                    for (partition_id, node_id) in flow_part2node_id {
-                        let key =
-                            FlownodeFlowKey::new(*node_id, *flow_id, *partition_id).to_bytes();
-                        keys.push(key);
-                        let key = FlowRouteKey::new(*flow_id, *partition_id).to_bytes();
-                        keys.push(key);
-                    }
-
-                    for key in keys {
-                        self.invalidate_key(&key).await;
-                    }
-                }
                 CacheIdent::FlowName(FlowName {
                     catalog_name,
                     flow_name,

src/common/meta/src/ddl/alter_database.rs

@@ -21,7 +21,7 @@ use snafu::{ensure, ResultExt};
 use strum::AsRefStr;
 
 use crate::cache_invalidator::Context;
-use crate::ddl::utils::map_to_procedure_error;
+use crate::ddl::utils::handle_retry_error;
 use crate::ddl::DdlContext;
 use crate::error::{Result, SchemaNotFoundSnafu};
 use crate::instruction::CacheIdent;
@@ -148,7 +148,7 @@ impl Procedure for AlterDatabaseProcedure {
             AlterDatabaseState::UpdateMetadata => self.on_update_metadata().await,
             AlterDatabaseState::InvalidateSchemaCache => self.on_invalidate_schema_cache().await,
         }
-        .map_err(map_to_procedure_error)
+        .map_err(handle_retry_error)
     }
 
     fn dump(&self) -> ProcedureResult<String> {

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

@@ -32,12 +32,9 @@ use store_api::metric_engine_consts::ALTER_PHYSICAL_EXTENSION_KEY;
 use strum::AsRefStr;
 use table::metadata::TableId;
 
-use crate::ddl::utils::{
-    add_peer_context_if_needed, map_to_procedure_error, sync_follower_regions,
-};
+use crate::ddl::utils::{add_peer_context_if_needed, sync_follower_regions};
 use crate::ddl::DdlContext;
-use crate::error::{DecodeJsonSnafu, MetadataCorruptionSnafu, Result};
-use crate::instruction::CacheIdent;
+use crate::error::{DecodeJsonSnafu, Error, MetadataCorruptionSnafu, Result};
 use crate::key::table_info::TableInfoValue;
 use crate::key::table_route::PhysicalTableRouteValue;
 use crate::key::DeserializedValueWithBytes;
@@ -69,7 +66,6 @@ impl AlterLogicalTablesProcedure {
                 physical_table_info: None,
                 physical_table_route: None,
                 physical_columns: vec![],
-                table_cache_keys_to_invalidate: vec![],
             },
         }
     }
@@ -199,19 +195,16 @@ impl AlterLogicalTablesProcedure {
         self.update_physical_table_metadata().await?;
         self.update_logical_tables_metadata().await?;
 
-        self.data.build_cache_keys_to_invalidate();
-        self.data.clear_metadata_fields();
-
         self.data.state = AlterTablesState::InvalidateTableCache;
         Ok(Status::executing(true))
     }
 
     pub(crate) async fn on_invalidate_table_cache(&mut self) -> Result<Status> {
-        let to_invalidate = &self.data.table_cache_keys_to_invalidate;
+        let to_invalidate = self.build_table_cache_keys_to_invalidate();
 
         self.context
             .cache_invalidator
-            .invalidate(&Default::default(), to_invalidate)
+            .invalidate(&Default::default(), &to_invalidate)
             .await?;
         Ok(Status::done())
     }
@@ -224,6 +217,14 @@ impl Procedure for AlterLogicalTablesProcedure {
     }
 
     async fn execute(&mut self, _ctx: &Context) -> ProcedureResult<Status> {
+        let error_handler = |e: Error| {
+            if e.is_retry_later() {
+                common_procedure::Error::retry_later(e)
+            } else {
+                common_procedure::Error::external(e)
+            }
+        };
+
         let state = &self.data.state;
 
         let step = state.as_ref();
@@ -240,7 +241,7 @@ impl Procedure for AlterLogicalTablesProcedure {
             AlterTablesState::UpdateMetadata => self.on_update_metadata().await,
             AlterTablesState::InvalidateTableCache => self.on_invalidate_table_cache().await,
         }
-        .map_err(map_to_procedure_error)
+        .map_err(error_handler)
     }
 
     fn dump(&self) -> ProcedureResult<String> {
@@ -279,20 +280,6 @@ pub struct AlterTablesData {
     physical_table_info: Option<DeserializedValueWithBytes<TableInfoValue>>,
     physical_table_route: Option<PhysicalTableRouteValue>,
     physical_columns: Vec<ColumnMetadata>,
-    table_cache_keys_to_invalidate: Vec<CacheIdent>,
-}
-
-impl AlterTablesData {
-    /// Clears all data fields except `state` and `table_cache_keys_to_invalidate` after metadata update.
-    /// This is done to avoid persisting unnecessary data after the update metadata step.
-    fn clear_metadata_fields(&mut self) {
-        self.tasks.clear();
-        self.table_info_values.clear();
-        self.physical_table_id = 0;
-        self.physical_table_info = None;
-        self.physical_table_route = None;
-        self.physical_columns.clear();
-    }
 }
 
 #[derive(Debug, Serialize, Deserialize, AsRefStr)]

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

@@ -15,12 +15,13 @@
 use table::metadata::RawTableInfo;
 use table::table_name::TableName;
 
-use crate::ddl::alter_logical_tables::AlterTablesData;
+use crate::ddl::alter_logical_tables::AlterLogicalTablesProcedure;
 use crate::instruction::CacheIdent;
 
-impl AlterTablesData {
-    pub(crate) fn build_cache_keys_to_invalidate(&mut self) {
+impl AlterLogicalTablesProcedure {
+    pub(crate) fn build_table_cache_keys_to_invalidate(&self) -> Vec<CacheIdent> {
         let mut cache_keys = self
+            .data
             .table_info_values
             .iter()
             .flat_map(|table| {
@@ -30,14 +31,14 @@ impl AlterTablesData {
                 ]
             })
             .collect::<Vec<_>>();
-        cache_keys.push(CacheIdent::TableId(self.physical_table_id));
+        cache_keys.push(CacheIdent::TableId(self.data.physical_table_id));
         // Safety: physical_table_info already filled in previous steps
-        let physical_table_info = &self.physical_table_info.as_ref().unwrap().table_info;
+        let physical_table_info = &self.data.physical_table_info.as_ref().unwrap().table_info;
         cache_keys.push(CacheIdent::TableName(extract_table_name(
             physical_table_info,
         )));
 
-        self.table_cache_keys_to_invalidate = cache_keys;
+        cache_keys
     }
 }
 

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

@@ -40,11 +40,10 @@ use table::table_reference::TableReference;
 
 use crate::cache_invalidator::Context;
 use crate::ddl::utils::{
-    add_peer_context_if_needed, handle_multiple_results, map_to_procedure_error,
-    sync_follower_regions, MultipleResults,
+    add_peer_context_if_needed, handle_multiple_results, sync_follower_regions, MultipleResults,
 };
 use crate::ddl::DdlContext;
-use crate::error::{AbortProcedureSnafu, NoLeaderSnafu, PutPoisonSnafu, Result, RetryLaterSnafu};
+use crate::error::{AbortProcedureSnafu, Error, NoLeaderSnafu, PutPoisonSnafu, Result};
 use crate::instruction::CacheIdent;
 use crate::key::table_info::TableInfoValue;
 use crate::key::{DeserializedValueWithBytes, RegionDistribution};
@@ -196,10 +195,7 @@ impl AlterTableProcedure {
             }
             MultipleResults::AllRetryable(error) => {
                 // Just returns the error, and wait for the next try.
-                let err = BoxedError::new(error);
-                Err(err).context(RetryLaterSnafu {
-                    clean_poisons: true,
-                })
+                Err(error)
             }
             MultipleResults::Ok(results) => {
                 self.submit_sync_region_requests(results, &physical_table_route.region_routes)
@@ -327,6 +323,16 @@ impl Procedure for AlterTableProcedure {
     }
 
     async fn execute(&mut self, ctx: &ProcedureContext) -> ProcedureResult<Status> {
+        let error_handler = |e: Error| {
+            if e.is_retry_later() {
+                ProcedureError::retry_later(e)
+            } else if e.need_clean_poisons() {
+                ProcedureError::external_and_clean_poisons(e)
+            } else {
+                ProcedureError::external(e)
+            }
+        };
+
         let state = &self.data.state;
 
         let step = state.as_ref();
@@ -344,7 +350,7 @@ impl Procedure for AlterTableProcedure {
             AlterTableState::UpdateMetadata => self.on_update_metadata().await,
             AlterTableState::InvalidateTableCache => self.on_broadcast().await,
         }
-        .map_err(map_to_procedure_error)
+        .map_err(error_handler)
     }
 
     fn dump(&self) -> ProcedureResult<String> {

src/common/meta/src/ddl/create_database.rs

@@ -22,7 +22,7 @@ use serde_with::{serde_as, DefaultOnNull};
 use snafu::{ensure, ResultExt};
 use strum::AsRefStr;
 
-use crate::ddl::utils::map_to_procedure_error;
+use crate::ddl::utils::handle_retry_error;
 use crate::ddl::DdlContext;
 use crate::error::{self, Result};
 use crate::key::schema_name::{SchemaNameKey, SchemaNameValue};
@@ -115,7 +115,7 @@ impl Procedure for CreateDatabaseProcedure {
             CreateDatabaseState::Prepare => self.on_prepare().await,
             CreateDatabaseState::CreateMetadata => self.on_create_metadata().await,
         }
-        .map_err(map_to_procedure_error)
+        .map_err(handle_retry_error)
     }
 
     fn dump(&self) -> ProcedureResult<String> {

src/common/meta/src/ddl/create_flow.rs

@@ -36,7 +36,7 @@ use strum::AsRefStr;
 use table::metadata::TableId;
 
 use crate::cache_invalidator::Context;
-use crate::ddl::utils::{add_peer_context_if_needed, map_to_procedure_error};
+use crate::ddl::utils::{add_peer_context_if_needed, handle_retry_error};
 use crate::ddl::DdlContext;
 use crate::error::{self, Result, UnexpectedSnafu};
 use crate::instruction::{CacheIdent, CreateFlow, DropFlow};
@@ -304,7 +304,7 @@ impl Procedure for CreateFlowProcedure {
             CreateFlowState::CreateMetadata => self.on_create_metadata().await,
             CreateFlowState::InvalidateFlowCache => self.on_broadcast().await,
         }
-        .map_err(map_to_procedure_error)
+        .map_err(handle_retry_error)
     }
 
     fn dump(&self) -> ProcedureResult<String> {

src/common/meta/src/ddl/create_logical_tables.rs

@@ -33,9 +33,7 @@ use store_api::storage::{RegionId, RegionNumber};
 use strum::AsRefStr;
 use table::metadata::{RawTableInfo, TableId};
 
-use crate::ddl::utils::{
-    add_peer_context_if_needed, map_to_procedure_error, sync_follower_regions,
-};
+use crate::ddl::utils::{add_peer_context_if_needed, handle_retry_error, sync_follower_regions};
 use crate::ddl::DdlContext;
 use crate::error::{DecodeJsonSnafu, MetadataCorruptionSnafu, Result};
 use crate::key::table_route::TableRouteValue;
@@ -240,7 +238,7 @@ impl Procedure for CreateLogicalTablesProcedure {
             CreateTablesState::DatanodeCreateRegions => self.on_datanode_create_regions().await,
             CreateTablesState::CreateMetadata => self.on_create_metadata().await,
         }
-        .map_err(map_to_procedure_error)
+        .map_err(handle_retry_error)
     }
 
     fn dump(&self) -> ProcedureResult<String> {

src/common/meta/src/ddl/create_table.rs

@@ -34,7 +34,7 @@ use table::table_reference::TableReference;
 
 use crate::ddl::create_table_template::{build_template, CreateRequestBuilder};
 use crate::ddl::utils::{
-    add_peer_context_if_needed, convert_region_routes_to_detecting_regions, map_to_procedure_error,
+    add_peer_context_if_needed, convert_region_routes_to_detecting_regions, handle_retry_error,
     region_storage_path,
 };
 use crate::ddl::{DdlContext, TableMetadata};
@@ -319,7 +319,7 @@ impl Procedure for CreateTableProcedure {
             CreateTableState::DatanodeCreateRegions => self.on_datanode_create_regions().await,
             CreateTableState::CreateMetadata => self.on_create_metadata().await,
         }
-        .map_err(map_to_procedure_error)
+        .map_err(handle_retry_error)
     }
 
     fn dump(&self) -> ProcedureResult<String> {

src/common/meta/src/ddl/create_view.rs

@@ -23,7 +23,7 @@ use table::metadata::{RawTableInfo, TableId, TableType};
 use table::table_reference::TableReference;
 
 use crate::cache_invalidator::Context;
-use crate::ddl::utils::map_to_procedure_error;
+use crate::ddl::utils::handle_retry_error;
 use crate::ddl::{DdlContext, TableMetadata};
 use crate::error::{self, Result};
 use crate::instruction::CacheIdent;
@@ -249,7 +249,7 @@ impl Procedure for CreateViewProcedure {
             CreateViewState::Prepare => self.on_prepare().await,
             CreateViewState::CreateMetadata => self.on_create_metadata(ctx).await,
         }
-        .map_err(map_to_procedure_error)
+        .map_err(handle_retry_error)
     }
 
     fn dump(&self) -> ProcedureResult<String> {

src/common/meta/src/ddl/drop_database.rs

@@ -21,7 +21,7 @@ use std::any::Any;
 use std::fmt::Debug;
 
 use common_error::ext::BoxedError;
-use common_procedure::error::{ExternalSnafu, FromJsonSnafu, ToJsonSnafu};
+use common_procedure::error::{Error as ProcedureError, ExternalSnafu, FromJsonSnafu, ToJsonSnafu};
 use common_procedure::{
     Context as ProcedureContext, LockKey, Procedure, Result as ProcedureResult, Status,
 };
@@ -31,7 +31,6 @@ use snafu::ResultExt;
 use tonic::async_trait;
 
 use self::start::DropDatabaseStart;
-use crate::ddl::utils::map_to_procedure_error;
 use crate::ddl::DdlContext;
 use crate::error::Result;
 use crate::key::table_name::TableNameValue;
@@ -142,7 +141,13 @@ impl Procedure for DropDatabaseProcedure {
         let (next, status) = state
             .next(&self.runtime_context, &mut self.context)
             .await
-            .map_err(map_to_procedure_error)?;
+            .map_err(|e| {
+                if e.is_retry_later() {
+                    ProcedureError::retry_later(e)
+                } else {
+                    ProcedureError::external(e)
+                }
+            })?;
 
         *state = next;
         Ok(status)

src/common/meta/src/ddl/drop_database/executor.rs

@@ -323,7 +323,6 @@ mod tests {
                     }
                     .build(),
                 ),
-                clean_poisons: false,
             })
         }
 

src/common/meta/src/ddl/drop_flow.rs

@@ -30,7 +30,7 @@ use snafu::{ensure, ResultExt};
 use strum::AsRefStr;
 
 use crate::cache_invalidator::Context;
-use crate::ddl::utils::{add_peer_context_if_needed, map_to_procedure_error};
+use crate::ddl::utils::{add_peer_context_if_needed, handle_retry_error};
 use crate::ddl::DdlContext;
 use crate::error::{self, Result};
 use crate::flow_name::FlowName;
@@ -201,7 +201,7 @@ impl Procedure for DropFlowProcedure {
             DropFlowState::InvalidateFlowCache => self.on_broadcast().await,
             DropFlowState::DropFlows => self.on_flownode_drop_flows().await,
         }
-        .map_err(map_to_procedure_error)
+        .map_err(handle_retry_error)
     }
 
     fn dump(&self) -> ProcedureResult<String> {

src/common/meta/src/ddl/drop_table.rs

@@ -35,7 +35,7 @@ use table::metadata::TableId;
 use table::table_reference::TableReference;
 
 use self::executor::DropTableExecutor;
-use crate::ddl::utils::map_to_procedure_error;
+use crate::ddl::utils::handle_retry_error;
 use crate::ddl::DdlContext;
 use crate::error::{self, Result};
 use crate::key::table_route::TableRouteValue;
@@ -221,7 +221,7 @@ impl Procedure for DropTableProcedure {
             DropTableState::DatanodeDropRegions => self.on_datanode_drop_regions().await,
             DropTableState::DeleteTombstone => self.on_delete_metadata_tombstone().await,
         }
-        .map_err(map_to_procedure_error)
+        .map_err(handle_retry_error)
     }
 
     fn dump(&self) -> ProcedureResult<String> {

src/common/meta/src/ddl/drop_view.rs

@@ -25,7 +25,7 @@ use table::metadata::{RawTableInfo, TableId, TableType};
 use table::table_reference::TableReference;
 
 use crate::cache_invalidator::Context;
-use crate::ddl::utils::map_to_procedure_error;
+use crate::ddl::utils::handle_retry_error;
 use crate::ddl::DdlContext;
 use crate::error::{self, Result};
 use crate::instruction::CacheIdent;
@@ -191,7 +191,7 @@ impl Procedure for DropViewProcedure {
             DropViewState::DeleteMetadata => self.on_delete_metadata().await,
             DropViewState::InvalidateViewCache => self.on_broadcast().await,
         }
-        .map_err(map_to_procedure_error)
+        .map_err(handle_retry_error)
     }
 
     fn dump(&self) -> ProcedureResult<String> {

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

@@ -105,7 +105,6 @@ impl MockDatanodeHandler for RetryErrorDatanodeHandler {
                 }
                 .build(),
             ),
-            clean_poisons: false,
         })
     }
 
@@ -219,7 +218,6 @@ impl MockDatanodeHandler for PartialSuccessDatanodeHandler {
                     }
                     .build(),
                 ),
-                clean_poisons: false,
             })
         } else {
             error::UnexpectedSnafu {
@@ -254,7 +252,6 @@ impl MockDatanodeHandler for AllFailureDatanodeHandler {
                     }
                     .build(),
                 ),
-                clean_poisons: false,
             })
         } else {
             error::UnexpectedSnafu {

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

@@ -575,7 +575,6 @@ async fn test_on_submit_alter_request_with_partial_success_retryable() {
         .await
         .unwrap_err();
     assert!(result.is_retry_later());
-    assert!(!result.need_clean_poisons());
 
     // Submits again
     let result = procedure
@@ -583,7 +582,6 @@ async fn test_on_submit_alter_request_with_partial_success_retryable() {
         .await
         .unwrap_err();
     assert!(result.is_retry_later());
-    assert!(!result.need_clean_poisons());
 }
 
 #[tokio::test]
@@ -620,14 +618,12 @@ async fn test_on_submit_alter_request_with_all_failure_retrybale() {
         .await
         .unwrap_err();
     assert!(err.is_retry_later());
-    assert!(err.need_clean_poisons());
     // submits again
     let err = procedure
         .submit_alter_region_requests(procedure_id, provider.as_ref())
         .await
         .unwrap_err();
     assert!(err.is_retry_later());
-    assert!(err.need_clean_poisons());
 }
 
 #[tokio::test]

src/common/meta/src/ddl/truncate_table.rs

@@ -31,7 +31,7 @@ use table::metadata::{RawTableInfo, TableId};
 use table::table_name::TableName;
 use table::table_reference::TableReference;
 
-use crate::ddl::utils::{add_peer_context_if_needed, map_to_procedure_error};
+use crate::ddl::utils::{add_peer_context_if_needed, handle_retry_error};
 use crate::ddl::DdlContext;
 use crate::error::{Result, TableNotFoundSnafu};
 use crate::key::table_info::TableInfoValue;
@@ -66,7 +66,7 @@ impl Procedure for TruncateTableProcedure {
                 self.on_datanode_truncate_regions().await
             }
         }
-        .map_err(map_to_procedure_error)
+        .map_err(handle_retry_error)
     }
 
     fn dump(&self) -> ProcedureResult<String> {

src/common/meta/src/ddl/utils.rs

@@ -60,16 +60,11 @@ pub fn add_peer_context_if_needed(datanode: Peer) -> impl FnOnce(Error) -> Error
     }
 }
 
-/// Maps the error to the corresponding procedure error.
-///
-/// This function determines whether the error should be retried and if poison cleanup is needed,
-/// then maps it to the appropriate procedure error variant.
-pub fn map_to_procedure_error(e: Error) -> ProcedureError {
-    match (e.is_retry_later(), e.need_clean_poisons()) {
-        (true, true) => ProcedureError::retry_later_and_clean_poisons(e),
-        (true, false) => ProcedureError::retry_later(e),
-        (false, true) => ProcedureError::external_and_clean_poisons(e),
-        (false, false) => ProcedureError::external(e),
+pub fn handle_retry_error(e: Error) -> ProcedureError {
+    if e.is_retry_later() {
+        ProcedureError::retry_later(e)
+    } else {
+        ProcedureError::external(e)
     }
 }
 

src/common/meta/src/ddl_manager.rs

@@ -47,10 +47,6 @@ use crate::error::{
 use crate::key::table_info::TableInfoValue;
 use crate::key::table_name::TableNameKey;
 use crate::key::{DeserializedValueWithBytes, TableMetadataManagerRef};
-#[cfg(feature = "enterprise")]
-use crate::rpc::ddl::trigger::CreateTriggerTask;
-#[cfg(feature = "enterprise")]
-use crate::rpc::ddl::DdlTask::CreateTrigger;
 use crate::rpc::ddl::DdlTask::{
     AlterDatabase, AlterLogicalTables, AlterTable, CreateDatabase, CreateFlow, CreateLogicalTables,
     CreateTable, CreateView, DropDatabase, DropFlow, DropLogicalTables, DropTable, DropView,
@@ -74,29 +70,8 @@ pub type BoxedProcedureLoaderFactory = dyn Fn(DdlContext) -> BoxedProcedureLoade
 pub struct DdlManager {
     ddl_context: DdlContext,
     procedure_manager: ProcedureManagerRef,
-    #[cfg(feature = "enterprise")]
-    trigger_ddl_manager: Option<TriggerDdlManagerRef>,
 }
 
-/// This trait is responsible for handling DDL tasks about triggers. e.g.,
-/// create trigger, drop trigger, etc.
-#[cfg(feature = "enterprise")]
-#[async_trait::async_trait]
-pub trait TriggerDdlManager: Send + Sync {
-    async fn create_trigger(
-        &self,
-        create_trigger_task: CreateTriggerTask,
-        procedure_manager: ProcedureManagerRef,
-        ddl_context: DdlContext,
-        query_context: QueryContext,
-    ) -> Result<SubmitDdlTaskResponse>;
-
-    fn as_any(&self) -> &dyn std::any::Any;
-}
-
-#[cfg(feature = "enterprise")]
-pub type TriggerDdlManagerRef = Arc<dyn TriggerDdlManager>;
-
 macro_rules! procedure_loader_entry {
     ($procedure:ident) => {
         (
@@ -125,13 +100,10 @@ impl DdlManager {
         ddl_context: DdlContext,
         procedure_manager: ProcedureManagerRef,
         register_loaders: bool,
-        #[cfg(feature = "enterprise")] trigger_ddl_manager: Option<TriggerDdlManagerRef>,
     ) -> Result<Self> {
         let manager = Self {
             ddl_context,
             procedure_manager,
-            #[cfg(feature = "enterprise")]
-            trigger_ddl_manager,
         };
         if register_loaders {
             manager.register_loaders()?;
@@ -697,28 +669,6 @@ async fn handle_create_flow_task(
     })
 }
 
-#[cfg(feature = "enterprise")]
-async fn handle_create_trigger_task(
-    ddl_manager: &DdlManager,
-    create_trigger_task: CreateTriggerTask,
-    query_context: QueryContext,
-) -> Result<SubmitDdlTaskResponse> {
-    let Some(m) = ddl_manager.trigger_ddl_manager.as_ref() else {
-        return UnsupportedSnafu {
-            operation: "create trigger",
-        }
-        .fail();
-    };
-
-    m.create_trigger(
-        create_trigger_task,
-        ddl_manager.procedure_manager.clone(),
-        ddl_manager.ddl_context.clone(),
-        query_context,
-    )
-    .await
-}
-
 async fn handle_alter_logical_table_tasks(
     ddl_manager: &DdlManager,
     alter_table_tasks: Vec<AlterTableTask>,
@@ -827,15 +777,6 @@ impl ProcedureExecutor for DdlManager {
                     handle_create_flow_task(self, create_flow_task, request.query_context.into())
                         .await
                 }
-                #[cfg(feature = "enterprise")]
-                CreateTrigger(create_trigger_task) => {
-                    handle_create_trigger_task(
-                        self,
-                        create_trigger_task,
-                        request.query_context.into(),
-                    )
-                    .await
-                }
                 DropFlow(drop_flow_task) => handle_drop_flow_task(self, drop_flow_task).await,
                 CreateView(create_view_task) => {
                     handle_create_view_task(self, create_view_task).await
@@ -964,8 +905,6 @@ mod tests {
             },
             procedure_manager.clone(),
             true,
-            #[cfg(feature = "enterprise")]
-            None,
         );
 
         let expected_loaders = vec![

src/common/meta/src/error.rs

@@ -454,10 +454,7 @@ pub enum Error {
     },
 
     #[snafu(display("Retry later"))]
-    RetryLater {
-        source: BoxedError,
-        clean_poisons: bool,
-    },
+    RetryLater { source: BoxedError },
 
     #[snafu(display("Abort procedure"))]
     AbortProcedure {
@@ -815,68 +812,6 @@ pub enum Error {
         #[snafu(source)]
         error: common_time::error::Error,
     },
-    #[snafu(display("Invalid file path: {}", file_path))]
-    InvalidFilePath {
-        #[snafu(implicit)]
-        location: Location,
-        file_path: String,
-    },
-
-    #[snafu(display("Failed to serialize flexbuffers"))]
-    SerializeFlexbuffers {
-        #[snafu(implicit)]
-        location: Location,
-        #[snafu(source)]
-        error: flexbuffers::SerializationError,
-    },
-
-    #[snafu(display("Failed to deserialize flexbuffers"))]
-    DeserializeFlexbuffers {
-        #[snafu(implicit)]
-        location: Location,
-        #[snafu(source)]
-        error: flexbuffers::DeserializationError,
-    },
-
-    #[snafu(display("Failed to read flexbuffers"))]
-    ReadFlexbuffers {
-        #[snafu(implicit)]
-        location: Location,
-        #[snafu(source)]
-        error: flexbuffers::ReaderError,
-    },
-
-    #[snafu(display("Invalid file name: {}", reason))]
-    InvalidFileName {
-        #[snafu(implicit)]
-        location: Location,
-        reason: String,
-    },
-
-    #[snafu(display("Invalid file extension: {}", reason))]
-    InvalidFileExtension {
-        #[snafu(implicit)]
-        location: Location,
-        reason: String,
-    },
-
-    #[snafu(display("Failed to write object, file path: {}", file_path))]
-    WriteObject {
-        #[snafu(implicit)]
-        location: Location,
-        file_path: String,
-        #[snafu(source)]
-        error: object_store::Error,
-    },
-
-    #[snafu(display("Failed to read object, file path: {}", file_path))]
-    ReadObject {
-        #[snafu(implicit)]
-        location: Location,
-        file_path: String,
-        #[snafu(source)]
-        error: object_store::Error,
-    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -899,7 +834,6 @@ impl ErrorExt for Error {
             ValueNotExist { .. } | ProcedurePoisonConflict { .. } => StatusCode::Unexpected,
 
             Unsupported { .. } => StatusCode::Unsupported,
-            WriteObject { .. } | ReadObject { .. } => StatusCode::StorageUnavailable,
 
             SerdeJson { .. }
             | ParseOption { .. }
@@ -933,10 +867,7 @@ impl ErrorExt for Error {
             | FromUtf8 { .. }
             | MetadataCorruption { .. }
             | ParseWalOptions { .. }
-            | KafkaGetOffset { .. }
-            | ReadFlexbuffers { .. }
-            | SerializeFlexbuffers { .. }
-            | DeserializeFlexbuffers { .. } => StatusCode::Unexpected,
+            | KafkaGetOffset { .. } => StatusCode::Unexpected,
 
             SendMessage { .. } | GetKvCache { .. } | CacheNotGet { .. } => StatusCode::Internal,
 
@@ -953,10 +884,7 @@ impl ErrorExt for Error {
             | InvalidSetDatabaseOption { .. }
             | InvalidUnsetDatabaseOption { .. }
             | InvalidTopicNamePrefix { .. }
-            | InvalidTimeZone { .. }
-            | InvalidFileExtension { .. }
-            | InvalidFileName { .. }
-            | InvalidFilePath { .. } => StatusCode::InvalidArguments,
+            | InvalidTimeZone { .. } => StatusCode::InvalidArguments,
             InvalidFlowRequestBody { .. } => StatusCode::InvalidArguments,
 
             FlowNotFound { .. } => StatusCode::FlowNotFound,
@@ -1042,7 +970,6 @@ impl Error {
     pub fn retry_later<E: ErrorExt + Send + Sync + 'static>(err: E) -> Error {
         Error::RetryLater {
             source: BoxedError::new(err),
-            clean_poisons: false,
         }
     }
 
@@ -1053,13 +980,7 @@ impl Error {
 
     /// Determine whether it needs to clean poisons.
     pub fn need_clean_poisons(&self) -> bool {
-        matches!(
-            self,
-            Error::AbortProcedure { clean_poisons, .. } if *clean_poisons
-        ) || matches!(
-            self,
-            Error::RetryLater { clean_poisons, .. } if *clean_poisons
-        )
+        matches!(self, Error::AbortProcedure { clean_poisons, .. } if *clean_poisons)
     }
 
     /// Returns true if the response exceeds the size limit.

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

@@ -45,7 +45,7 @@ use crate::kv_backend::KvBackendRef;
 use crate::rpc::store::BatchDeleteRequest;
 
 /// The key of `__flow/` scope.
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, PartialEq)]
 pub struct FlowScoped<T> {
     inner: T,
 }

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

@@ -153,15 +153,6 @@ impl FlowInfoValue {
         &self.flownode_ids
     }
 
-    /// Insert a new flownode id for a partition.
-    pub fn insert_flownode_id(
-        &mut self,
-        partition: FlowPartitionId,
-        node: FlownodeId,
-    ) -> Option<FlownodeId> {
-        self.flownode_ids.insert(partition, node)
-    }
-
     /// Returns the `source_table`.
     pub fn source_table_ids(&self) -> &[TableId] {
         &self.source_table_ids

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

@@ -205,7 +205,7 @@ impl FlowNameManager {
         catalog: &str,
     ) -> BoxStream<'static, Result<(String, FlowNameValue)>> {
         let start_key = FlowNameKey::range_start_key(catalog);
-        common_telemetry::trace!("flow_names: start_key: {:?}", start_key);
+        common_telemetry::debug!("flow_names: start_key: {:?}", start_key);
         let req = RangeRequest::new().with_prefix(start_key);
 
         let stream = PaginationStream::new(

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

@@ -42,7 +42,7 @@ lazy_static! {
 /// The key stores the route info of the flow.
 ///
 /// The layout: `__flow/route/{flow_id}/{partition_id}`.
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, PartialEq)]
 pub struct FlowRouteKey(FlowScoped<FlowRouteKeyInner>);
 
 impl FlowRouteKey {
@@ -145,12 +145,6 @@ pub struct FlowRouteValue {
     pub(crate) peer: Peer,
 }
 
-impl From<Peer> for FlowRouteValue {
-    fn from(peer: Peer) -> Self {
-        Self { peer }
-    }
-}
-
 impl FlowRouteValue {
     /// Returns the `peer`.
     pub fn peer(&self) -> &Peer {

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

@@ -166,17 +166,6 @@ impl FlownodeFlowManager {
         Self { kv_backend }
     }
 
-    /// Whether given flow exist on this flownode.
-    pub async fn exists(
-        &self,
-        flownode_id: FlownodeId,
-        flow_id: FlowId,
-        partition_id: FlowPartitionId,
-    ) -> Result<bool> {
-        let key = FlownodeFlowKey::new(flownode_id, flow_id, partition_id).to_bytes();
-        Ok(self.kv_backend.get(&key).await?.is_some())
-    }
-
     /// Retrieves all [FlowId] and [FlowPartitionId]s of the specified `flownode_id`.
     pub fn flows(
         &self,

src/common/meta/src/kv_backend.rs

@@ -38,14 +38,6 @@ pub mod txn;
 pub mod util;
 pub type KvBackendRef<E = Error> = Arc<dyn KvBackend<Error = E> + Send + Sync>;
 
-#[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
-// The default meta table name, default is "greptime_metakv".
-pub const DEFAULT_META_TABLE_NAME: &str = "greptime_metakv";
-
-#[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
-// The default lock id for election, default is 1.
-pub const DEFAULT_META_ELECTION_LOCK_ID: u64 = 1;
-
 #[async_trait]
 pub trait KvBackend: TxnService
 where

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

@@ -308,11 +308,10 @@ mod tests {
     use super::*;
     use crate::error::Error;
     use crate::kv_backend::test::{
-        prepare_kv, prepare_kv_with_prefix, test_kv_batch_delete, test_kv_batch_get,
-        test_kv_compare_and_put, test_kv_delete_range, test_kv_put, test_kv_range, test_kv_range_2,
-        test_simple_kv_range, test_txn_compare_equal, test_txn_compare_greater,
-        test_txn_compare_less, test_txn_compare_not_equal, test_txn_one_compare_op,
-        text_txn_multi_compare_op, unprepare_kv,
+        prepare_kv, test_kv_batch_delete, test_kv_batch_get, test_kv_compare_and_put,
+        test_kv_delete_range, test_kv_put, test_kv_range, test_kv_range_2, test_txn_compare_equal,
+        test_txn_compare_greater, test_txn_compare_less, test_txn_compare_not_equal,
+        test_txn_one_compare_op, text_txn_multi_compare_op,
     };
 
     async fn mock_mem_store_with_data() -> MemoryKvBackend<Error> {
@@ -381,12 +380,4 @@ mod tests {
         test_txn_compare_less(&kv_backend).await;
         test_txn_compare_not_equal(&kv_backend).await;
     }
-    #[tokio::test]
-    async fn test_mem_all_range() {
-        let kv_backend = MemoryKvBackend::<Error>::new();
-        let prefix = b"";
-        prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
-        test_simple_kv_range(&kv_backend).await;
-        unprepare_kv(&kv_backend, prefix).await;
-    }
 }

src/common/meta/src/kv_backend/rds.rs

@@ -33,12 +33,6 @@ use crate::rpc::store::{
 };
 use crate::rpc::KeyValue;
 
-const RDS_STORE_OP_BATCH_GET: &str = "batch_get";
-const RDS_STORE_OP_BATCH_PUT: &str = "batch_put";
-const RDS_STORE_OP_RANGE_QUERY: &str = "range_query";
-const RDS_STORE_OP_RANGE_DELETE: &str = "range_delete";
-const RDS_STORE_OP_BATCH_DELETE: &str = "batch_delete";
-
 #[cfg(feature = "pg_kvbackend")]
 mod postgres;
 #[cfg(feature = "pg_kvbackend")]
@@ -566,21 +560,3 @@ fn check_txn_ops(txn_ops: &[TxnOp]) -> Result<bool> {
     });
     Ok(same)
 }
-
-#[macro_export]
-macro_rules! record_rds_sql_execute_elapsed {
-    ($result:expr, $label_store:expr,$label_op:expr,$label_type:expr) => {{
-        let timer = std::time::Instant::now();
-        $result
-            .inspect(|_| {
-                $crate::metrics::RDS_SQL_EXECUTE_ELAPSED
-                    .with_label_values(&[$label_store, "success", $label_op, $label_type])
-                    .observe(timer.elapsed().as_millis_f64())
-            })
-            .inspect_err(|_| {
-                $crate::metrics::RDS_SQL_EXECUTE_ELAPSED
-                    .with_label_values(&[$label_store, "error", $label_op, $label_type])
-                    .observe(timer.elapsed().as_millis_f64());
-            })
-    }};
-}

src/common/meta/src/kv_backend/rds/mysql.rs

@@ -20,13 +20,11 @@ use snafu::ResultExt;
 use sqlx::mysql::MySqlRow;
 use sqlx::pool::Pool;
 use sqlx::{MySql, MySqlPool, Row, Transaction as MySqlTransaction};
-use strum::AsRefStr;
 
 use crate::error::{CreateMySqlPoolSnafu, MySqlExecutionSnafu, MySqlTransactionSnafu, Result};
 use crate::kv_backend::rds::{
     Executor, ExecutorFactory, ExecutorImpl, KvQueryExecutor, RdsStore, Transaction,
-    RDS_STORE_OP_BATCH_DELETE, RDS_STORE_OP_BATCH_GET, RDS_STORE_OP_BATCH_PUT,
-    RDS_STORE_OP_RANGE_DELETE, RDS_STORE_OP_RANGE_QUERY, RDS_STORE_TXN_RETRY_COUNT,
+    RDS_STORE_TXN_RETRY_COUNT,
 };
 use crate::kv_backend::KvBackendRef;
 use crate::rpc::store::{
@@ -35,8 +33,6 @@ use crate::rpc::store::{
 };
 use crate::rpc::KeyValue;
 
-const MYSQL_STORE_NAME: &str = "mysql_store";
-
 type MySqlClient = Arc<Pool<MySql>>;
 pub struct MySqlTxnClient(MySqlTransaction<'static, MySql>);
 
@@ -51,7 +47,7 @@ fn key_value_from_row(row: MySqlRow) -> KeyValue {
 const EMPTY: &[u8] = &[0];
 
 /// Type of range template.
-#[derive(Debug, Clone, Copy, AsRefStr)]
+#[derive(Debug, Clone, Copy)]
 enum RangeTemplateType {
     Point,
     Range,
@@ -62,8 +58,6 @@ enum RangeTemplateType {
 
 /// Builds params for the given range template type.
 impl RangeTemplateType {
-    /// Builds the parameters for the given range template type.
-    /// You can check out the conventions at [RangeRequest]
     fn build_params(&self, mut key: Vec<u8>, range_end: Vec<u8>) -> Vec<Vec<u8>> {
         match self {
             RangeTemplateType::Point => vec![key],
@@ -166,7 +160,7 @@ impl<'a> MySqlTemplateFactory<'a> {
             range_template: RangeTemplate {
                 point: format!("SELECT k, v FROM `{table_name}` WHERE k = ?"),
                 range: format!("SELECT k, v FROM `{table_name}` WHERE k >= ? AND k < ? ORDER BY k"),
-                full: format!("SELECT k, v FROM `{table_name}` ORDER BY k"),
+                full: format!("SELECT k, v FROM `{table_name}` ? ORDER BY k"),
                 left_bounded: format!("SELECT k, v FROM `{table_name}` WHERE k >= ? ORDER BY k"),
                 prefix: format!("SELECT k, v FROM `{table_name}` WHERE k LIKE ? ORDER BY k"),
             },
@@ -349,12 +343,7 @@ impl KvQueryExecutor<MySqlClient> for MySqlStore {
             RangeTemplate::with_limit(template, if req.limit == 0 { 0 } else { req.limit + 1 });
         let limit = req.limit as usize;
         debug!("query: {:?}, params: {:?}", query, params);
-        let mut kvs = crate::record_rds_sql_execute_elapsed!(
-            query_executor.query(&query, &params_ref).await,
-            MYSQL_STORE_NAME,
-            RDS_STORE_OP_RANGE_QUERY,
-            template_type.as_ref()
-        )?;
+        let mut kvs = query_executor.query(&query, &params_ref).await?;
         if req.keys_only {
             kvs.iter_mut().for_each(|kv| kv.value = vec![]);
         }
@@ -392,12 +381,7 @@ impl KvQueryExecutor<MySqlClient> for MySqlStore {
 
         // Fast path: if we don't need previous kvs, we can just upsert the keys.
         if !req.prev_kv {
-            crate::record_rds_sql_execute_elapsed!(
-                query_executor.execute(&update, &values_params).await,
-                MYSQL_STORE_NAME,
-                RDS_STORE_OP_BATCH_PUT,
-                ""
-            )?;
+            query_executor.execute(&update, &values_params).await?;
             return Ok(BatchPutResponse::default());
         }
         // Should use transaction to ensure atomicity.
@@ -408,12 +392,7 @@ impl KvQueryExecutor<MySqlClient> for MySqlStore {
             txn.commit().await?;
             return res;
         }
-        let prev_kvs = crate::record_rds_sql_execute_elapsed!(
-            query_executor.query(&select, &in_params).await,
-            MYSQL_STORE_NAME,
-            RDS_STORE_OP_BATCH_PUT,
-            ""
-        )?;
+        let prev_kvs = query_executor.query(&select, &in_params).await?;
         query_executor.execute(&update, &values_params).await?;
         Ok(BatchPutResponse { prev_kvs })
     }
@@ -430,12 +409,7 @@ impl KvQueryExecutor<MySqlClient> for MySqlStore {
             .sql_template_set
             .generate_batch_get_query(req.keys.len());
         let params = req.keys.iter().map(|x| x as _).collect::<Vec<_>>();
-        let kvs = crate::record_rds_sql_execute_elapsed!(
-            query_executor.query(&query, &params).await,
-            MYSQL_STORE_NAME,
-            RDS_STORE_OP_BATCH_GET,
-            ""
-        )?;
+        let kvs = query_executor.query(&query, &params).await?;
         Ok(BatchGetResponse { kvs })
     }
 
@@ -467,12 +441,7 @@ impl KvQueryExecutor<MySqlClient> for MySqlStore {
         let template = self.sql_template_set.delete_template.get(template_type);
         let params = template_type.build_params(req.key, req.range_end);
         let params_ref = params.iter().map(|x| x as _).collect::<Vec<_>>();
-        crate::record_rds_sql_execute_elapsed!(
-            query_executor.execute(template, &params_ref).await,
-            MYSQL_STORE_NAME,
-            RDS_STORE_OP_RANGE_DELETE,
-            template_type.as_ref()
-        )?;
+        query_executor.execute(template, &params_ref).await?;
         let mut resp = DeleteRangeResponse::new(prev_kvs.len() as i64);
         if req.prev_kv {
             resp.with_prev_kvs(prev_kvs);
@@ -494,12 +463,7 @@ impl KvQueryExecutor<MySqlClient> for MySqlStore {
         let params = req.keys.iter().map(|x| x as _).collect::<Vec<_>>();
         // Fast path: if we don't need previous kvs, we can just delete the keys.
         if !req.prev_kv {
-            crate::record_rds_sql_execute_elapsed!(
-                query_executor.execute(&query, &params).await,
-                MYSQL_STORE_NAME,
-                RDS_STORE_OP_BATCH_DELETE,
-                ""
-            )?;
+            query_executor.execute(&query, &params).await?;
             return Ok(BatchDeleteResponse::default());
         }
         // Should use transaction to ensure atomicity.
@@ -519,12 +483,7 @@ impl KvQueryExecutor<MySqlClient> for MySqlStore {
             .await?
             .kvs;
         // Pure `DELETE` has no return value, so we need to use `execute` instead of `query`.
-        crate::record_rds_sql_execute_elapsed!(
-            query_executor.execute(&query, &params).await,
-            MYSQL_STORE_NAME,
-            RDS_STORE_OP_BATCH_DELETE,
-            ""
-        )?;
+        query_executor.execute(&query, &params).await?;
         if req.prev_kv {
             Ok(BatchDeleteResponse { prev_kvs })
         } else {
@@ -579,11 +538,10 @@ mod tests {
         prepare_kv_with_prefix, test_kv_batch_delete_with_prefix, test_kv_batch_get_with_prefix,
         test_kv_compare_and_put_with_prefix, test_kv_delete_range_with_prefix,
         test_kv_put_with_prefix, test_kv_range_2_with_prefix, test_kv_range_with_prefix,
-        test_simple_kv_range, test_txn_compare_equal, test_txn_compare_greater,
-        test_txn_compare_less, test_txn_compare_not_equal, test_txn_one_compare_op,
-        text_txn_multi_compare_op, unprepare_kv,
+        test_txn_compare_equal, test_txn_compare_greater, test_txn_compare_less,
+        test_txn_compare_not_equal, test_txn_one_compare_op, text_txn_multi_compare_op,
+        unprepare_kv,
     };
-    use crate::maybe_skip_mysql_integration_test;
 
     async fn build_mysql_kv_backend(table_name: &str) -> Option<MySqlStore> {
         init_default_ut_logging();
@@ -610,7 +568,6 @@ mod tests {
 
     #[tokio::test]
     async fn test_mysql_put() {
-        maybe_skip_mysql_integration_test!();
         let kv_backend = build_mysql_kv_backend("put_test").await.unwrap();
         let prefix = b"put/";
         prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
@@ -620,7 +577,6 @@ mod tests {
 
     #[tokio::test]
     async fn test_mysql_range() {
-        maybe_skip_mysql_integration_test!();
         let kv_backend = build_mysql_kv_backend("range_test").await.unwrap();
         let prefix = b"range/";
         prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
@@ -630,26 +586,14 @@ mod tests {
 
     #[tokio::test]
     async fn test_mysql_range_2() {
-        maybe_skip_mysql_integration_test!();
         let kv_backend = build_mysql_kv_backend("range2_test").await.unwrap();
         let prefix = b"range2/";
         test_kv_range_2_with_prefix(&kv_backend, prefix.to_vec()).await;
         unprepare_kv(&kv_backend, prefix).await;
     }
 
-    #[tokio::test]
-    async fn test_mysql_all_range() {
-        maybe_skip_mysql_integration_test!();
-        let kv_backend = build_mysql_kv_backend("simple_range_test").await.unwrap();
-        let prefix = b"";
-        prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
-        test_simple_kv_range(&kv_backend).await;
-        unprepare_kv(&kv_backend, prefix).await;
-    }
-
     #[tokio::test]
     async fn test_mysql_batch_get() {
-        maybe_skip_mysql_integration_test!();
         let kv_backend = build_mysql_kv_backend("batch_get_test").await.unwrap();
         let prefix = b"batch_get/";
         prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
@@ -659,7 +603,6 @@ mod tests {
 
     #[tokio::test]
     async fn test_mysql_batch_delete() {
-        maybe_skip_mysql_integration_test!();
         let kv_backend = build_mysql_kv_backend("batch_delete_test").await.unwrap();
         let prefix = b"batch_delete/";
         prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
@@ -669,7 +612,6 @@ mod tests {
 
     #[tokio::test]
     async fn test_mysql_batch_delete_with_prefix() {
-        maybe_skip_mysql_integration_test!();
         let kv_backend = build_mysql_kv_backend("batch_delete_with_prefix_test")
             .await
             .unwrap();
@@ -681,7 +623,6 @@ mod tests {
 
     #[tokio::test]
     async fn test_mysql_delete_range() {
-        maybe_skip_mysql_integration_test!();
         let kv_backend = build_mysql_kv_backend("delete_range_test").await.unwrap();
         let prefix = b"delete_range/";
         prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
@@ -691,7 +632,6 @@ mod tests {
 
     #[tokio::test]
     async fn test_mysql_compare_and_put() {
-        maybe_skip_mysql_integration_test!();
         let kv_backend = build_mysql_kv_backend("compare_and_put_test")
             .await
             .unwrap();
@@ -702,7 +642,6 @@ mod tests {
 
     #[tokio::test]
     async fn test_mysql_txn() {
-        maybe_skip_mysql_integration_test!();
         let kv_backend = build_mysql_kv_backend("txn_test").await.unwrap();
         test_txn_one_compare_op(&kv_backend).await;
         text_txn_multi_compare_op(&kv_backend).await;

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

@@ -18,7 +18,6 @@ use std::sync::Arc;
 use common_telemetry::debug;
 use deadpool_postgres::{Config, Pool, Runtime};
 use snafu::ResultExt;
-use strum::AsRefStr;
 use tokio_postgres::types::ToSql;
 use tokio_postgres::{IsolationLevel, NoTls, Row};
 
@@ -28,8 +27,7 @@ use crate::error::{
 };
 use crate::kv_backend::rds::{
     Executor, ExecutorFactory, ExecutorImpl, KvQueryExecutor, RdsStore, Transaction,
-    RDS_STORE_OP_BATCH_DELETE, RDS_STORE_OP_BATCH_GET, RDS_STORE_OP_BATCH_PUT,
-    RDS_STORE_OP_RANGE_DELETE, RDS_STORE_OP_RANGE_QUERY, RDS_STORE_TXN_RETRY_COUNT,
+    RDS_STORE_TXN_RETRY_COUNT,
 };
 use crate::kv_backend::KvBackendRef;
 use crate::rpc::store::{
@@ -38,8 +36,6 @@ use crate::rpc::store::{
 };
 use crate::rpc::KeyValue;
 
-const PG_STORE_NAME: &str = "pg_store";
-
 pub struct PgClient(deadpool::managed::Object<deadpool_postgres::Manager>);
 pub struct PgTxnClient<'a>(deadpool_postgres::Transaction<'a>);
 
@@ -54,7 +50,7 @@ fn key_value_from_row(r: Row) -> KeyValue {
 const EMPTY: &[u8] = &[0];
 
 /// Type of range template.
-#[derive(Debug, Clone, Copy, AsRefStr)]
+#[derive(Debug, Clone, Copy)]
 enum RangeTemplateType {
     Point,
     Range,
@@ -65,8 +61,6 @@ enum RangeTemplateType {
 
 /// Builds params for the given range template type.
 impl RangeTemplateType {
-    /// Builds the parameters for the given range template type.
-    /// You can check out the conventions at [RangeRequest]
     fn build_params(&self, mut key: Vec<u8>, range_end: Vec<u8>) -> Vec<Vec<u8>> {
         match self {
             RangeTemplateType::Point => vec![key],
@@ -170,7 +164,7 @@ impl<'a> PgSqlTemplateFactory<'a> {
                 range: format!(
                     "SELECT k, v FROM \"{table_name}\" WHERE k >= $1 AND k < $2 ORDER BY k"
                 ),
-                full: format!("SELECT k, v FROM \"{table_name}\" ORDER BY k"),
+                full: format!("SELECT k, v FROM \"{table_name}\" $1 ORDER BY k"),
                 left_bounded: format!("SELECT k, v FROM \"{table_name}\" WHERE k >= $1 ORDER BY k"),
                 prefix: format!("SELECT k, v FROM \"{table_name}\" WHERE k LIKE $1 ORDER BY k"),
             },
@@ -364,13 +358,7 @@ impl KvQueryExecutor<PgClient> for PgStore {
             RangeTemplate::with_limit(template, if req.limit == 0 { 0 } else { req.limit + 1 });
         let limit = req.limit as usize;
         debug!("query: {:?}, params: {:?}", query, params);
-        let mut kvs = crate::record_rds_sql_execute_elapsed!(
-            query_executor.query(&query, &params_ref).await,
-            PG_STORE_NAME,
-            RDS_STORE_OP_RANGE_QUERY,
-            template_type.as_ref()
-        )?;
-
+        let mut kvs = query_executor.query(&query, &params_ref).await?;
         if req.keys_only {
             kvs.iter_mut().for_each(|kv| kv.value = vec![]);
         }
@@ -405,13 +393,7 @@ impl KvQueryExecutor<PgClient> for PgStore {
         let query = self
             .sql_template_set
             .generate_batch_upsert_query(req.kvs.len());
-
-        let kvs = crate::record_rds_sql_execute_elapsed!(
-            query_executor.query(&query, &params).await,
-            PG_STORE_NAME,
-            RDS_STORE_OP_BATCH_PUT,
-            ""
-        )?;
+        let kvs = query_executor.query(&query, &params).await?;
         if req.prev_kv {
             Ok(BatchPutResponse { prev_kvs: kvs })
         } else {
@@ -432,12 +414,7 @@ impl KvQueryExecutor<PgClient> for PgStore {
             .sql_template_set
             .generate_batch_get_query(req.keys.len());
         let params = req.keys.iter().map(|x| x as _).collect::<Vec<_>>();
-        let kvs = crate::record_rds_sql_execute_elapsed!(
-            query_executor.query(&query, &params).await,
-            PG_STORE_NAME,
-            RDS_STORE_OP_BATCH_GET,
-            ""
-        )?;
+        let kvs = query_executor.query(&query, &params).await?;
         Ok(BatchGetResponse { kvs })
     }
 
@@ -450,12 +427,7 @@ impl KvQueryExecutor<PgClient> for PgStore {
         let template = self.sql_template_set.delete_template.get(template_type);
         let params = template_type.build_params(req.key, req.range_end);
         let params_ref = params.iter().map(|x| x as _).collect::<Vec<_>>();
-        let kvs = crate::record_rds_sql_execute_elapsed!(
-            query_executor.query(template, &params_ref).await,
-            PG_STORE_NAME,
-            RDS_STORE_OP_RANGE_DELETE,
-            template_type.as_ref()
-        )?;
+        let kvs = query_executor.query(template, &params_ref).await?;
         let mut resp = DeleteRangeResponse::new(kvs.len() as i64);
         if req.prev_kv {
             resp.with_prev_kvs(kvs);
@@ -475,13 +447,7 @@ impl KvQueryExecutor<PgClient> for PgStore {
             .sql_template_set
             .generate_batch_delete_query(req.keys.len());
         let params = req.keys.iter().map(|x| x as _).collect::<Vec<_>>();
-
-        let kvs = crate::record_rds_sql_execute_elapsed!(
-            query_executor.query(&query, &params).await,
-            PG_STORE_NAME,
-            RDS_STORE_OP_BATCH_DELETE,
-            ""
-        )?;
+        let kvs = query_executor.query(&query, &params).await?;
         if req.prev_kv {
             Ok(BatchDeleteResponse { prev_kvs: kvs })
         } else {
@@ -545,11 +511,10 @@ mod tests {
         prepare_kv_with_prefix, test_kv_batch_delete_with_prefix, test_kv_batch_get_with_prefix,
         test_kv_compare_and_put_with_prefix, test_kv_delete_range_with_prefix,
         test_kv_put_with_prefix, test_kv_range_2_with_prefix, test_kv_range_with_prefix,
-        test_simple_kv_range, test_txn_compare_equal, test_txn_compare_greater,
-        test_txn_compare_less, test_txn_compare_not_equal, test_txn_one_compare_op,
-        text_txn_multi_compare_op, unprepare_kv,
+        test_txn_compare_equal, test_txn_compare_greater, test_txn_compare_less,
+        test_txn_compare_not_equal, test_txn_one_compare_op, text_txn_multi_compare_op,
+        unprepare_kv,
     };
-    use crate::maybe_skip_postgres_integration_test;
 
     async fn build_pg_kv_backend(table_name: &str) -> Option<PgStore> {
         let endpoints = std::env::var("GT_POSTGRES_ENDPOINTS").unwrap_or_default();
@@ -584,7 +549,6 @@ mod tests {
 
     #[tokio::test]
     async fn test_pg_put() {
-        maybe_skip_postgres_integration_test!();
         let kv_backend = build_pg_kv_backend("put_test").await.unwrap();
         let prefix = b"put/";
         prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
@@ -594,7 +558,6 @@ mod tests {
 
     #[tokio::test]
     async fn test_pg_range() {
-        maybe_skip_postgres_integration_test!();
         let kv_backend = build_pg_kv_backend("range_test").await.unwrap();
         let prefix = b"range/";
         prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
@@ -604,26 +567,14 @@ mod tests {
 
     #[tokio::test]
     async fn test_pg_range_2() {
-        maybe_skip_postgres_integration_test!();
         let kv_backend = build_pg_kv_backend("range2_test").await.unwrap();
         let prefix = b"range2/";
         test_kv_range_2_with_prefix(&kv_backend, prefix.to_vec()).await;
         unprepare_kv(&kv_backend, prefix).await;
     }
 
-    #[tokio::test]
-    async fn test_pg_all_range() {
-        maybe_skip_postgres_integration_test!();
-        let kv_backend = build_pg_kv_backend("simple_range_test").await.unwrap();
-        let prefix = b"";
-        prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
-        test_simple_kv_range(&kv_backend).await;
-        unprepare_kv(&kv_backend, prefix).await;
-    }
-
     #[tokio::test]
     async fn test_pg_batch_get() {
-        maybe_skip_postgres_integration_test!();
         let kv_backend = build_pg_kv_backend("batch_get_test").await.unwrap();
         let prefix = b"batch_get/";
         prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
@@ -633,7 +584,6 @@ mod tests {
 
     #[tokio::test]
     async fn test_pg_batch_delete() {
-        maybe_skip_postgres_integration_test!();
         let kv_backend = build_pg_kv_backend("batch_delete_test").await.unwrap();
         let prefix = b"batch_delete/";
         prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
@@ -643,7 +593,6 @@ mod tests {
 
     #[tokio::test]
     async fn test_pg_batch_delete_with_prefix() {
-        maybe_skip_postgres_integration_test!();
         let kv_backend = build_pg_kv_backend("batch_delete_with_prefix_test")
             .await
             .unwrap();
@@ -655,7 +604,6 @@ mod tests {
 
     #[tokio::test]
     async fn test_pg_delete_range() {
-        maybe_skip_postgres_integration_test!();
         let kv_backend = build_pg_kv_backend("delete_range_test").await.unwrap();
         let prefix = b"delete_range/";
         prepare_kv_with_prefix(&kv_backend, prefix.to_vec()).await;
@@ -665,7 +613,6 @@ mod tests {
 
     #[tokio::test]
     async fn test_pg_compare_and_put() {
-        maybe_skip_postgres_integration_test!();
         let kv_backend = build_pg_kv_backend("compare_and_put_test").await.unwrap();
         let prefix = b"compare_and_put/";
         let kv_backend = Arc::new(kv_backend);
@@ -674,7 +621,6 @@ mod tests {
 
     #[tokio::test]
     async fn test_pg_txn() {
-        maybe_skip_postgres_integration_test!();
         let kv_backend = build_pg_kv_backend("txn_test").await.unwrap();
         test_txn_one_compare_op(&kv_backend).await;
         text_txn_multi_compare_op(&kv_backend).await;

src/common/meta/src/kv_backend/test.rs

@@ -108,44 +108,6 @@ pub async fn test_kv_range(kv_backend: &impl KvBackend) {
     test_kv_range_with_prefix(kv_backend, vec![]).await;
 }
 
-pub async fn test_simple_kv_range(kvbackend: &impl KvBackend) {
-    {
-        let full_query = RangeRequest::new().with_range(vec![0], vec![0]);
-        let response = kvbackend.range(full_query).await.unwrap();
-        assert_eq!(response.kvs.len(), 4);
-    }
-    {
-        let point_query = RangeRequest::new().with_range(b"key11".to_vec(), vec![]);
-        let response = kvbackend.range(point_query).await.unwrap();
-        assert_eq!(response.kvs.len(), 1);
-    }
-    {
-        let left_bounded_query = RangeRequest::new().with_range(b"key1".to_vec(), vec![0]);
-        let response = kvbackend.range(left_bounded_query).await.unwrap();
-        assert_eq!(response.kvs.len(), 4);
-    }
-    {
-        let range_query = RangeRequest::new().with_range(b"key1".to_vec(), b"key11".to_vec());
-        let response = kvbackend.range(range_query).await.unwrap();
-        assert_eq!(response.kvs.len(), 1);
-    }
-    {
-        let prefix_query = RangeRequest::new().with_range(b"key1".to_vec(), b"key2".to_vec());
-        let response = kvbackend.range(prefix_query).await.unwrap();
-        assert_eq!(response.kvs.len(), 2);
-    }
-    {
-        let range_query = RangeRequest::new().with_range(b"key10".to_vec(), b"key100".to_vec());
-        let response = kvbackend.range(range_query).await.unwrap();
-        assert_eq!(response.kvs.len(), 0);
-    }
-    {
-        let prefix_query = RangeRequest::new().with_range(b"key10".to_vec(), b"key11".to_vec());
-        let response = kvbackend.range(prefix_query).await.unwrap();
-        assert_eq!(response.kvs.len(), 0);
-    }
-}
-
 pub async fn test_kv_range_with_prefix(kv_backend: &impl KvBackend, prefix: Vec<u8>) {
     let key = [prefix.clone(), b"key1".to_vec()].concat();
     let key11 = [prefix.clone(), b"key11".to_vec()].concat();

src/common/meta/src/lib.rs

@@ -15,7 +15,6 @@
 #![feature(assert_matches)]
 #![feature(btree_extract_if)]
 #![feature(let_chains)]
-#![feature(duration_millis_float)]
 
 pub mod cache;
 pub mod cache_invalidator;
@@ -42,7 +41,6 @@ pub mod region_keeper;
 pub mod region_registry;
 pub mod rpc;
 pub mod sequence;
-pub mod snapshot;
 pub mod state_store;
 #[cfg(any(test, feature = "testing"))]
 pub mod test_util;

src/common/meta/src/metrics.rs

@@ -108,10 +108,4 @@ lazy_static! {
         &["name"]
     )
     .unwrap();
-    pub static ref RDS_SQL_EXECUTE_ELAPSED: HistogramVec = register_histogram_vec!(
-        "greptime_meta_rds_pg_sql_execute_elapsed_ms",
-        "rds pg sql execute elapsed",
-        &["backend", "result", "op", "type"]
-    )
-    .unwrap();
 }

src/common/meta/src/rpc/ddl.rs

@@ -12,9 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#[cfg(feature = "enterprise")]
-pub mod trigger;
-
 use std::collections::{HashMap, HashSet};
 use std::result;
 
@@ -71,8 +68,6 @@ pub enum DdlTask {
     DropFlow(DropFlowTask),
     CreateView(CreateViewTask),
     DropView(DropViewTask),
-    #[cfg(feature = "enterprise")]
-    CreateTrigger(trigger::CreateTriggerTask),
 }
 
 impl DdlTask {
@@ -247,18 +242,6 @@ impl TryFrom<Task> for DdlTask {
             Task::DropFlowTask(drop_flow) => Ok(DdlTask::DropFlow(drop_flow.try_into()?)),
             Task::CreateViewTask(create_view) => Ok(DdlTask::CreateView(create_view.try_into()?)),
             Task::DropViewTask(drop_view) => Ok(DdlTask::DropView(drop_view.try_into()?)),
-            Task::CreateTriggerTask(create_trigger) => {
-                #[cfg(feature = "enterprise")]
-                return Ok(DdlTask::CreateTrigger(create_trigger.try_into()?));
-                #[cfg(not(feature = "enterprise"))]
-                {
-                    let _ = create_trigger;
-                    crate::error::UnsupportedSnafu {
-                        operation: "create trigger",
-                    }
-                    .fail()
-                }
-            }
         }
     }
 }
@@ -309,8 +292,6 @@ impl TryFrom<SubmitDdlTaskRequest> for PbDdlTaskRequest {
             DdlTask::DropFlow(task) => Task::DropFlowTask(task.into()),
             DdlTask::CreateView(task) => Task::CreateViewTask(task.try_into()?),
             DdlTask::DropView(task) => Task::DropViewTask(task.into()),
-            #[cfg(feature = "enterprise")]
-            DdlTask::CreateTrigger(task) => Task::CreateTriggerTask(task.into()),
         };
 
         Ok(Self {

src/common/meta/src/rpc/ddl/trigger.rs

@@ -1,276 +0,0 @@
-use std::collections::HashMap;
-use std::time::Duration;
-
-use api::v1::meta::CreateTriggerTask as PbCreateTriggerTask;
-use api::v1::notify_channel::ChannelType as PbChannelType;
-use api::v1::{
-    CreateTriggerExpr, NotifyChannel as PbNotifyChannel, WebhookOptions as PbWebhookOptions,
-};
-use serde::{Deserialize, Serialize};
-use snafu::OptionExt;
-
-use crate::error;
-use crate::error::Result;
-use crate::rpc::ddl::DdlTask;
-
-// Create trigger
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct CreateTriggerTask {
-    pub catalog_name: String,
-    pub trigger_name: String,
-    pub if_not_exists: bool,
-    pub sql: String,
-    pub channels: Vec<NotifyChannel>,
-    pub labels: HashMap<String, String>,
-    pub annotations: HashMap<String, String>,
-    pub interval: Duration,
-}
-
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-pub struct NotifyChannel {
-    pub name: String,
-    pub channel_type: ChannelType,
-}
-
-/// The available channel enum for sending trigger notifications.
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-pub enum ChannelType {
-    Webhook(WebhookOptions),
-}
-
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-pub struct WebhookOptions {
-    /// The URL of the AlertManager API endpoint.
-    ///
-    /// e.g., "http://localhost:9093".
-    pub url: String,
-    /// Configuration options for the AlertManager webhook. e.g., timeout, etc.
-    pub opts: HashMap<String, String>,
-}
-
-impl From<CreateTriggerTask> for PbCreateTriggerTask {
-    fn from(task: CreateTriggerTask) -> Self {
-        let channels = task
-            .channels
-            .into_iter()
-            .map(PbNotifyChannel::from)
-            .collect();
-
-        let expr = CreateTriggerExpr {
-            catalog_name: task.catalog_name,
-            trigger_name: task.trigger_name,
-            create_if_not_exists: task.if_not_exists,
-            sql: task.sql,
-            channels,
-            labels: task.labels,
-            annotations: task.annotations,
-            interval: task.interval.as_secs(),
-        };
-
-        PbCreateTriggerTask {
-            create_trigger: Some(expr),
-        }
-    }
-}
-
-impl TryFrom<PbCreateTriggerTask> for CreateTriggerTask {
-    type Error = error::Error;
-
-    fn try_from(task: PbCreateTriggerTask) -> Result<Self> {
-        let expr = task.create_trigger.context(error::InvalidProtoMsgSnafu {
-            err_msg: "expected create_trigger",
-        })?;
-
-        let channels = expr
-            .channels
-            .into_iter()
-            .map(NotifyChannel::try_from)
-            .collect::<Result<Vec<_>>>()?;
-
-        let task = CreateTriggerTask {
-            catalog_name: expr.catalog_name,
-            trigger_name: expr.trigger_name,
-            if_not_exists: expr.create_if_not_exists,
-            sql: expr.sql,
-            channels,
-            labels: expr.labels,
-            annotations: expr.annotations,
-            interval: Duration::from_secs(expr.interval),
-        };
-        Ok(task)
-    }
-}
-
-impl From<NotifyChannel> for PbNotifyChannel {
-    fn from(channel: NotifyChannel) -> Self {
-        let NotifyChannel { name, channel_type } = channel;
-
-        let channel_type = match channel_type {
-            ChannelType::Webhook(options) => PbChannelType::Webhook(PbWebhookOptions {
-                url: options.url,
-                opts: options.opts,
-            }),
-        };
-
-        PbNotifyChannel {
-            name,
-            channel_type: Some(channel_type),
-        }
-    }
-}
-
-impl TryFrom<PbNotifyChannel> for NotifyChannel {
-    type Error = error::Error;
-
-    fn try_from(channel: PbNotifyChannel) -> Result<Self> {
-        let PbNotifyChannel { name, channel_type } = channel;
-
-        let channel_type = channel_type.context(error::InvalidProtoMsgSnafu {
-            err_msg: "expected channel_type",
-        })?;
-
-        let channel_type = match channel_type {
-            PbChannelType::Webhook(options) => ChannelType::Webhook(WebhookOptions {
-                url: options.url,
-                opts: options.opts,
-            }),
-        };
-        Ok(NotifyChannel { name, channel_type })
-    }
-}
-
-impl DdlTask {
-    /// Creates a [`DdlTask`] to create a trigger.
-    pub fn new_create_trigger(expr: CreateTriggerTask) -> Self {
-        DdlTask::CreateTrigger(expr)
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_convert_create_trigger_task() {
-        let original = CreateTriggerTask {
-            catalog_name: "test_catalog".to_string(),
-            trigger_name: "test_trigger".to_string(),
-            if_not_exists: true,
-            sql: "SELECT * FROM test".to_string(),
-            channels: vec![
-                NotifyChannel {
-                    name: "channel1".to_string(),
-                    channel_type: ChannelType::Webhook(WebhookOptions {
-                        url: "http://localhost:9093".to_string(),
-                        opts: HashMap::from([("timeout".to_string(), "30s".to_string())]),
-                    }),
-                },
-                NotifyChannel {
-                    name: "channel2".to_string(),
-                    channel_type: ChannelType::Webhook(WebhookOptions {
-                        url: "http://alertmanager:9093".to_string(),
-                        opts: HashMap::new(),
-                    }),
-                },
-            ],
-            labels: vec![
-                ("key1".to_string(), "value1".to_string()),
-                ("key2".to_string(), "value2".to_string()),
-            ]
-            .into_iter()
-            .collect(),
-            annotations: vec![
-                ("summary".to_string(), "Test alert".to_string()),
-                ("description".to_string(), "This is a test".to_string()),
-            ]
-            .into_iter()
-            .collect(),
-            interval: Duration::from_secs(60),
-        };
-
-        let pb_task: PbCreateTriggerTask = original.clone().into();
-
-        let expr = pb_task.create_trigger.as_ref().unwrap();
-        assert_eq!(expr.catalog_name, "test_catalog");
-        assert_eq!(expr.trigger_name, "test_trigger");
-        assert!(expr.create_if_not_exists);
-        assert_eq!(expr.sql, "SELECT * FROM test");
-        assert_eq!(expr.channels.len(), 2);
-        assert_eq!(expr.labels.len(), 2);
-        assert_eq!(expr.labels.get("key1").unwrap(), "value1");
-        assert_eq!(expr.labels.get("key2").unwrap(), "value2");
-        assert_eq!(expr.annotations.len(), 2);
-        assert_eq!(expr.annotations.get("summary").unwrap(), "Test alert");
-        assert_eq!(
-            expr.annotations.get("description").unwrap(),
-            "This is a test"
-        );
-        assert_eq!(expr.interval, 60);
-
-        let round_tripped = CreateTriggerTask::try_from(pb_task).unwrap();
-
-        assert_eq!(original.catalog_name, round_tripped.catalog_name);
-        assert_eq!(original.trigger_name, round_tripped.trigger_name);
-        assert_eq!(original.if_not_exists, round_tripped.if_not_exists);
-        assert_eq!(original.sql, round_tripped.sql);
-        assert_eq!(original.channels.len(), round_tripped.channels.len());
-        assert_eq!(&original.channels[0], &round_tripped.channels[0]);
-        assert_eq!(&original.channels[1], &round_tripped.channels[1]);
-        assert_eq!(original.labels, round_tripped.labels);
-        assert_eq!(original.annotations, round_tripped.annotations);
-        assert_eq!(original.interval, round_tripped.interval);
-
-        // Invalid, since create_trigger is None and it's required.
-        let invalid_task = PbCreateTriggerTask {
-            create_trigger: None,
-        };
-        let result = CreateTriggerTask::try_from(invalid_task);
-        assert!(result.is_err());
-    }
-
-    #[test]
-    fn test_convert_notify_channel() {
-        let original = NotifyChannel {
-            name: "test_channel".to_string(),
-            channel_type: ChannelType::Webhook(WebhookOptions {
-                url: "http://localhost:9093".to_string(),
-                opts: HashMap::new(),
-            }),
-        };
-        let pb_channel: PbNotifyChannel = original.clone().into();
-        match pb_channel.channel_type.as_ref().unwrap() {
-            PbChannelType::Webhook(options) => {
-                assert_eq!(pb_channel.name, "test_channel");
-                assert_eq!(options.url, "http://localhost:9093");
-                assert!(options.opts.is_empty());
-            }
-        }
-        let round_tripped = NotifyChannel::try_from(pb_channel).unwrap();
-        assert_eq!(original, round_tripped);
-
-        // Test with timeout is None.
-        let no_timeout = NotifyChannel {
-            name: "no_timeout".to_string(),
-            channel_type: ChannelType::Webhook(WebhookOptions {
-                url: "http://localhost:9093".to_string(),
-                opts: HashMap::new(),
-            }),
-        };
-        let pb_no_timeout: PbNotifyChannel = no_timeout.clone().into();
-        match pb_no_timeout.channel_type.as_ref().unwrap() {
-            PbChannelType::Webhook(options) => {
-                assert_eq!(options.url, "http://localhost:9093");
-            }
-        }
-        let round_tripped_no_timeout = NotifyChannel::try_from(pb_no_timeout).unwrap();
-        assert_eq!(no_timeout, round_tripped_no_timeout);
-
-        // Invalid, since channel_type is None and it's required.
-        let invalid_channel = PbNotifyChannel {
-            name: "invalid".to_string(),
-            channel_type: None,
-        };
-        let result = NotifyChannel::try_from(invalid_channel);
-        assert!(result.is_err());
-    }
-}

src/common/meta/src/snapshot.rs

@@ -1,380 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-pub mod file;
-
-use std::fmt::{Display, Formatter};
-use std::path::{Path, PathBuf};
-use std::time::Instant;
-
-use common_telemetry::info;
-use file::{Metadata, MetadataContent};
-use futures::TryStreamExt;
-use object_store::ObjectStore;
-use snafu::{OptionExt, ResultExt};
-use strum::Display;
-
-use crate::error::{
-    Error, InvalidFileExtensionSnafu, InvalidFileNameSnafu, InvalidFilePathSnafu, ReadObjectSnafu,
-    Result, WriteObjectSnafu,
-};
-use crate::kv_backend::KvBackendRef;
-use crate::range_stream::{PaginationStream, DEFAULT_PAGE_SIZE};
-use crate::rpc::store::{BatchPutRequest, RangeRequest};
-use crate::rpc::KeyValue;
-use crate::snapshot::file::{Document, KeyValue as FileKeyValue};
-
-/// The format of the backup file.
-#[derive(Debug, PartialEq, Eq, Display, Clone, Copy)]
-pub enum FileFormat {
-    #[strum(serialize = "fb")]
-    FlexBuffers,
-}
-
-impl TryFrom<&str> for FileFormat {
-    type Error = String;
-
-    fn try_from(value: &str) -> std::result::Result<Self, Self::Error> {
-        match value.to_lowercase().as_str() {
-            "fb" => Ok(FileFormat::FlexBuffers),
-            _ => Err(format!("Invalid file format: {}", value)),
-        }
-    }
-}
-
-#[derive(Debug, PartialEq, Eq, Display)]
-#[strum(serialize_all = "lowercase")]
-pub enum DataType {
-    Metadata,
-}
-
-impl TryFrom<&str> for DataType {
-    type Error = String;
-
-    fn try_from(value: &str) -> std::result::Result<Self, Self::Error> {
-        match value.to_lowercase().as_str() {
-            "metadata" => Ok(DataType::Metadata),
-            _ => Err(format!("Invalid data type: {}", value)),
-        }
-    }
-}
-
-#[derive(Debug, PartialEq, Eq)]
-pub struct FileExtension {
-    format: FileFormat,
-    data_type: DataType,
-}
-
-impl FileExtension {
-    pub fn new(format: FileFormat, data_type: DataType) -> Self {
-        Self { format, data_type }
-    }
-}
-
-impl Display for FileExtension {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}.{}", self.data_type, self.format)
-    }
-}
-
-impl TryFrom<&str> for FileExtension {
-    type Error = Error;
-
-    fn try_from(value: &str) -> Result<Self> {
-        let parts = value.split(".").collect::<Vec<&str>>();
-        if parts.len() != 2 {
-            return InvalidFileExtensionSnafu {
-                reason: format!(
-                    "Extension should be in the format of <datatype>.<format>, got: {}",
-                    value
-                ),
-            }
-            .fail();
-        }
-
-        let data_type = DataType::try_from(parts[0])
-            .map_err(|e| InvalidFileExtensionSnafu { reason: e }.build())?;
-        let format = FileFormat::try_from(parts[1])
-            .map_err(|e| InvalidFileExtensionSnafu { reason: e }.build())?;
-        Ok(FileExtension { format, data_type })
-    }
-}
-
-#[derive(Debug, PartialEq, Eq)]
-pub struct FileName {
-    name: String,
-    extension: FileExtension,
-}
-
-impl Display for FileName {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}.{}", self.name, self.extension)
-    }
-}
-
-impl TryFrom<&str> for FileName {
-    type Error = Error;
-
-    fn try_from(value: &str) -> Result<Self> {
-        let Some((name, extension)) = value.split_once(".") else {
-            return InvalidFileNameSnafu {
-                reason: format!(
-                    "The file name should be in the format of <name>.<extension>, got: {}",
-                    value
-                ),
-            }
-            .fail();
-        };
-        let extension = FileExtension::try_from(extension)?;
-        Ok(Self {
-            name: name.to_string(),
-            extension,
-        })
-    }
-}
-
-impl FileName {
-    fn new(name: String, extension: FileExtension) -> Self {
-        Self { name, extension }
-    }
-}
-
-/// The manager of the metadata snapshot.
-///
-/// It manages the metadata snapshot, including dumping and restoring.
-pub struct MetadataSnapshotManager {
-    kv_backend: KvBackendRef,
-    object_store: ObjectStore,
-}
-
-/// The maximum size of the request to put metadata, use 1MiB by default.
-const MAX_REQUEST_SIZE: usize = 1024 * 1024;
-
-impl MetadataSnapshotManager {
-    pub fn new(kv_backend: KvBackendRef, object_store: ObjectStore) -> Self {
-        Self {
-            kv_backend,
-            object_store,
-        }
-    }
-
-    /// Restores the metadata from the backup file to the metadata store.
-    pub async fn restore(&self, file_path: &str) -> Result<u64> {
-        let path = Path::new(file_path);
-
-        let file_name = path
-            .file_name()
-            .and_then(|s| s.to_str())
-            .context(InvalidFilePathSnafu { file_path })?;
-
-        let filename = FileName::try_from(file_name)?;
-        let data = self
-            .object_store
-            .read(file_path)
-            .await
-            .context(ReadObjectSnafu { file_path })?;
-        let document = Document::from_slice(&filename.extension.format, &data.to_bytes())?;
-        let metadata_content = document.into_metadata_content()?;
-        let mut req = BatchPutRequest::default();
-        let mut total_request_size = 0;
-        let mut count = 0;
-        let now = Instant::now();
-        for FileKeyValue { key, value } in metadata_content.into_iter() {
-            count += 1;
-            let key_size = key.len();
-            let value_size = value.len();
-            if total_request_size + key_size + value_size > MAX_REQUEST_SIZE {
-                self.kv_backend.batch_put(req).await?;
-                req = BatchPutRequest::default();
-                total_request_size = 0;
-            }
-            req.kvs.push(KeyValue { key, value });
-            total_request_size += key_size + value_size;
-        }
-        if !req.kvs.is_empty() {
-            self.kv_backend.batch_put(req).await?;
-        }
-
-        info!(
-            "Restored metadata from {} successfully, total {} key-value pairs, elapsed {:?}",
-            file_path,
-            count,
-            now.elapsed()
-        );
-        Ok(count)
-    }
-
-    pub async fn check_target_source_clean(&self) -> Result<bool> {
-        let req = RangeRequest::new().with_range(vec![0], vec![0]);
-        let mut stream = Box::pin(
-            PaginationStream::new(self.kv_backend.clone(), req, 1, Result::Ok).into_stream(),
-        );
-        let v = stream.as_mut().try_next().await?;
-        Ok(v.is_none())
-    }
-
-    /// Dumps the metadata to the backup file.
-    pub async fn dump(&self, path: &str, filename_str: &str) -> Result<(String, u64)> {
-        let format = FileFormat::FlexBuffers;
-        let filename = FileName::new(
-            filename_str.to_string(),
-            FileExtension {
-                format,
-                data_type: DataType::Metadata,
-            },
-        );
-        let file_path_buf = [path, filename.to_string().as_str()]
-            .iter()
-            .collect::<PathBuf>();
-        let file_path = file_path_buf.to_str().context(InvalidFileNameSnafu {
-            reason: format!("Invalid file path: {}, filename: {}", path, filename_str),
-        })?;
-        let now = Instant::now();
-        let req = RangeRequest::new().with_range(vec![0], vec![0]);
-        let stream = PaginationStream::new(self.kv_backend.clone(), req, DEFAULT_PAGE_SIZE, |kv| {
-            Ok(FileKeyValue {
-                key: kv.key,
-                value: kv.value,
-            })
-        })
-        .into_stream();
-        let keyvalues = stream.try_collect::<Vec<_>>().await?;
-        let num_keyvalues = keyvalues.len();
-        let document = Document::new(
-            Metadata::new(),
-            file::Content::Metadata(MetadataContent::new(keyvalues)),
-        );
-        let bytes = document.to_bytes(&format)?;
-        let r = self
-            .object_store
-            .write(file_path, bytes)
-            .await
-            .context(WriteObjectSnafu { file_path })?;
-        info!(
-            "Dumped metadata to {} successfully, total {} key-value pairs, file size {} bytes, elapsed {:?}",
-            file_path,
-            num_keyvalues,
-            r.content_length(),
-            now.elapsed()
-        );
-
-        Ok((filename.to_string(), num_keyvalues as u64))
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::assert_matches::assert_matches;
-    use std::sync::Arc;
-
-    use common_test_util::temp_dir::{create_temp_dir, TempDir};
-    use object_store::services::Fs;
-
-    use super::*;
-    use crate::kv_backend::memory::MemoryKvBackend;
-    use crate::kv_backend::KvBackend;
-    use crate::rpc::store::PutRequest;
-
-    #[test]
-    fn test_file_name() {
-        let file_name = FileName::try_from("test.metadata.fb").unwrap();
-        assert_eq!(file_name.name, "test");
-        assert_eq!(file_name.extension.format, FileFormat::FlexBuffers);
-        assert_eq!(file_name.extension.data_type, DataType::Metadata);
-        assert_eq!(file_name.to_string(), "test.metadata.fb");
-
-        let invalid_file_name = FileName::try_from("test.metadata").unwrap_err();
-        assert_eq!(
-            invalid_file_name.to_string(),
-            "Invalid file extension: Extension should be in the format of <datatype>.<format>, got: metadata"
-        );
-
-        let invalid_file_extension = FileName::try_from("test.metadata.hello").unwrap_err();
-        assert_eq!(
-            invalid_file_extension.to_string(),
-            "Invalid file extension: Invalid file format: hello"
-        );
-    }
-
-    fn test_env(
-        prefix: &str,
-    ) -> (
-        TempDir,
-        Arc<MemoryKvBackend<Error>>,
-        MetadataSnapshotManager,
-    ) {
-        let temp_dir = create_temp_dir(prefix);
-        let kv_backend = Arc::new(MemoryKvBackend::default());
-        let temp_path = temp_dir.path();
-        let data_path = temp_path.join("data").as_path().display().to_string();
-        let builder = Fs::default().root(&data_path);
-        let object_store = ObjectStore::new(builder).unwrap().finish();
-        let manager = MetadataSnapshotManager::new(kv_backend.clone(), object_store);
-        (temp_dir, kv_backend, manager)
-    }
-
-    #[tokio::test]
-    async fn test_dump_and_restore() {
-        common_telemetry::init_default_ut_logging();
-        let (temp_dir, kv_backend, manager) = test_env("test_dump_and_restore");
-        let temp_path = temp_dir.path();
-
-        for i in 0..10 {
-            kv_backend
-                .put(
-                    PutRequest::new()
-                        .with_key(format!("test_{}", i).as_bytes().to_vec())
-                        .with_value(format!("value_{}", i).as_bytes().to_vec()),
-                )
-                .await
-                .unwrap();
-        }
-        let dump_path = temp_path.join("snapshot");
-        manager
-            .dump(
-                &dump_path.as_path().display().to_string(),
-                "metadata_snapshot",
-            )
-            .await
-            .unwrap();
-        // Clean up the kv backend
-        kv_backend.clear();
-
-        let restore_path = dump_path
-            .join("metadata_snapshot.metadata.fb")
-            .as_path()
-            .display()
-            .to_string();
-        manager.restore(&restore_path).await.unwrap();
-
-        for i in 0..10 {
-            let key = format!("test_{}", i);
-            let value = kv_backend.get(key.as_bytes()).await.unwrap().unwrap();
-            assert_eq!(value.value, format!("value_{}", i).as_bytes());
-        }
-    }
-
-    #[tokio::test]
-    async fn test_restore_from_nonexistent_file() {
-        let (temp_dir, _kv_backend, manager) = test_env("test_restore_from_nonexistent_file");
-        let restore_path = temp_dir
-            .path()
-            .join("nonexistent.metadata.fb")
-            .as_path()
-            .display()
-            .to_string();
-        let err = manager.restore(&restore_path).await.unwrap_err();
-        assert_matches!(err, Error::ReadObject { .. })
-    }
-}

src/common/meta/src/snapshot/file.rs

@@ -1,145 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use common_time::util::current_time_millis;
-use flexbuffers::{FlexbufferSerializer, Reader};
-use serde::{Deserialize, Serialize};
-use snafu::ResultExt;
-
-use crate::error::{
-    DeserializeFlexbuffersSnafu, ReadFlexbuffersSnafu, Result, SerializeFlexbuffersSnafu,
-};
-use crate::snapshot::FileFormat;
-
-/// The layout of the backup file.
-#[derive(Debug, PartialEq, Serialize, Deserialize)]
-pub(crate) struct Document {
-    metadata: Metadata,
-    content: Content,
-}
-
-impl Document {
-    /// Creates a new document.
-    pub fn new(metadata: Metadata, content: Content) -> Self {
-        Self { metadata, content }
-    }
-
-    fn serialize_to_flexbuffer(&self) -> Result<Vec<u8>> {
-        let mut builder = FlexbufferSerializer::new();
-        self.serialize(&mut builder)
-            .context(SerializeFlexbuffersSnafu)?;
-        Ok(builder.take_buffer())
-    }
-
-    /// Converts the [`Document`] to a bytes.
-    pub(crate) fn to_bytes(&self, format: &FileFormat) -> Result<Vec<u8>> {
-        match format {
-            FileFormat::FlexBuffers => self.serialize_to_flexbuffer(),
-        }
-    }
-
-    fn deserialize_from_flexbuffer(data: &[u8]) -> Result<Self> {
-        let reader = Reader::get_root(data).context(ReadFlexbuffersSnafu)?;
-        Document::deserialize(reader).context(DeserializeFlexbuffersSnafu)
-    }
-
-    /// Deserializes the [`Document`] from a bytes.
-    pub(crate) fn from_slice(format: &FileFormat, data: &[u8]) -> Result<Self> {
-        match format {
-            FileFormat::FlexBuffers => Self::deserialize_from_flexbuffer(data),
-        }
-    }
-
-    /// Converts the [`Document`] to a [`MetadataContent`].
-    pub(crate) fn into_metadata_content(self) -> Result<MetadataContent> {
-        match self.content {
-            Content::Metadata(metadata) => Ok(metadata),
-        }
-    }
-}
-
-/// The metadata of the backup file.
-#[derive(Debug, PartialEq, Serialize, Deserialize)]
-pub(crate) struct Metadata {
-    // UNIX_EPOCH in milliseconds.
-    created_timestamp_mills: i64,
-}
-
-impl Metadata {
-    /// Create a new metadata.
-    ///
-    /// The `created_timestamp_mills` will be the current time in milliseconds.
-    pub fn new() -> Self {
-        Self {
-            created_timestamp_mills: current_time_millis(),
-        }
-    }
-}
-
-/// The content of the backup file.
-#[derive(Debug, PartialEq, Serialize, Deserialize)]
-pub(crate) enum Content {
-    Metadata(MetadataContent),
-}
-
-/// The content of the backup file.
-#[derive(Debug, PartialEq, Serialize, Deserialize)]
-pub(crate) struct MetadataContent {
-    values: Vec<KeyValue>,
-}
-
-impl MetadataContent {
-    /// Create a new metadata content.
-    pub fn new(values: impl IntoIterator<Item = KeyValue>) -> Self {
-        Self {
-            values: values.into_iter().collect(),
-        }
-    }
-
-    /// Returns an iterator over the key-value pairs.
-    pub fn into_iter(self) -> impl Iterator<Item = KeyValue> {
-        self.values.into_iter()
-    }
-}
-
-/// The key-value pair of the backup file.
-#[derive(Debug, PartialEq, Serialize, Deserialize)]
-pub(crate) struct KeyValue {
-    pub key: Vec<u8>,
-    pub value: Vec<u8>,
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_document() {
-        let document = Document::new(
-            Metadata::new(),
-            Content::Metadata(MetadataContent::new(vec![KeyValue {
-                key: b"key".to_vec(),
-                value: b"value".to_vec(),
-            }])),
-        );
-
-        let bytes = document.to_bytes(&FileFormat::FlexBuffers).unwrap();
-        let document_deserialized = Document::from_slice(&FileFormat::FlexBuffers, &bytes).unwrap();
-        assert_eq!(
-            document.metadata.created_timestamp_mills,
-            document_deserialized.metadata.created_timestamp_mills
-        );
-        assert_eq!(document.content, document_deserialized.content);
-    }
-}

src/common/meta/src/test_util.rs

@@ -233,35 +233,3 @@ pub async fn test_kafka_topic_pool(
 
     KafkaTopicPool::new(&config, kv_backend, topic_creator)
 }
-
-#[macro_export]
-/// Skip the test if the environment variable `GT_KAFKA_ENDPOINTS` is not set.
-///
-/// The format of the environment variable is:
-/// ```
-/// GT_KAFKA_ENDPOINTS=localhost:9092,localhost:9093
-/// ```
-macro_rules! maybe_skip_postgres_integration_test {
-    () => {
-        if std::env::var("GT_POSTGRES_ENDPOINTS").is_err() {
-            common_telemetry::warn!("The endpoints is empty, skipping the test");
-            return;
-        }
-    };
-}
-
-#[macro_export]
-/// Skip the test if the environment variable `GT_KAFKA_ENDPOINTS` is not set.
-///
-/// The format of the environment variable is:
-/// ```
-/// GT_KAFKA_ENDPOINTS=localhost:9092,localhost:9093
-/// ```
-macro_rules! maybe_skip_mysql_integration_test {
-    () => {
-        if std::env::var("GT_MYSQL_ENDPOINTS").is_err() {
-            common_telemetry::warn!("The endpoints is empty, skipping the test");
-            return;
-        }
-    };
-}

src/common/procedure/src/error.rs

@@ -138,10 +138,7 @@ pub enum Error {
     },
 
     #[snafu(display("Procedure exec failed"))]
-    RetryLater {
-        source: BoxedError,
-        clean_poisons: bool,
-    },
+    RetryLater { source: BoxedError },
 
     #[snafu(display("Procedure panics, procedure_id: {}", procedure_id))]
     ProcedurePanic { procedure_id: ProcedureId },
@@ -301,15 +298,6 @@ impl Error {
     pub fn retry_later<E: ErrorExt + Send + Sync + 'static>(err: E) -> Error {
         Error::RetryLater {
             source: BoxedError::new(err),
-            clean_poisons: false,
-        }
-    }
-
-    /// Creates a new [Error::RetryLater] error from source `err` and clean poisons.
-    pub fn retry_later_and_clean_poisons<E: ErrorExt + Send + Sync + 'static>(err: E) -> Error {
-        Error::RetryLater {
-            source: BoxedError::new(err),
-            clean_poisons: true,
         }
     }
 
@@ -321,7 +309,6 @@ impl Error {
     /// Determine whether it needs to clean poisons.
     pub fn need_clean_poisons(&self) -> bool {
         matches!(self, Error::External { clean_poisons, .. } if *clean_poisons)
-            || matches!(self, Error::RetryLater { clean_poisons, .. } if *clean_poisons)
     }
 
     /// Creates a new [Error::RetryLater] or [Error::External] error from source `err` according

src/common/procedure/src/local/runner.rs

@@ -358,11 +358,10 @@ impl Runner {
                     Err(e) => {
                         error!(
                             e;
-                            "Failed to execute procedure {}-{}, retry: {}, clean_poisons: {}",
+                            "Failed to execute procedure {}-{}, retry: {}",
                             self.procedure.type_name(),
                             self.meta.id,
                             e.is_retry_later(),
-                            e.need_clean_poisons(),
                         );
 
                         // Don't store state if `ProcedureManager` is stopped.
@@ -379,11 +378,6 @@ impl Runner {
                                 self.meta.set_state(ProcedureState::retrying(Arc::new(e)));
                                 return;
                             }
-                            debug!(
-                                "Procedure {}-{} cleaned poisons",
-                                self.procedure.type_name(),
-                                self.meta.id,
-                            );
                         }
 
                         if e.is_retry_later() {
@@ -587,7 +581,6 @@ impl Runner {
 
 #[cfg(test)]
 mod tests {
-    use std::assert_matches::assert_matches;
     use std::sync::atomic::{AtomicU64, Ordering};
     use std::sync::Arc;
 
@@ -1605,75 +1598,6 @@ mod tests {
         assert_eq!(&procedure_id.to_string(), ROOT_ID);
     }
 
-    #[tokio::test]
-    async fn test_execute_exceed_max_retry_after_set_poison() {
-        common_telemetry::init_default_ut_logging();
-        let mut times = 0;
-        let poison_key = PoisonKey::new("table/1024");
-        let moved_poison_key = poison_key.clone();
-        let exec_fn = move |ctx: Context| {
-            times += 1;
-            let poison_key = moved_poison_key.clone();
-            async move {
-                if times == 1 {
-                    Ok(Status::executing(true))
-                } else {
-                    // Put the poison to the context.
-                    ctx.provider
-                        .try_put_poison(&poison_key, ctx.procedure_id)
-                        .await
-                        .unwrap();
-                    Err(Error::retry_later_and_clean_poisons(MockError::new(
-                        StatusCode::Unexpected,
-                    )))
-                }
-            }
-            .boxed()
-        };
-        let poison = ProcedureAdapter {
-            data: "poison".to_string(),
-            lock_key: LockKey::single_exclusive("catalog.schema.table"),
-            poison_keys: PoisonKeys::new(vec![poison_key.clone()]),
-            exec_fn,
-            rollback_fn: None,
-        };
-
-        let dir = create_temp_dir("exceed_max_after_set_poison");
-        let meta = poison.new_meta(ROOT_ID);
-        let object_store = test_util::new_object_store(&dir);
-        let procedure_store = Arc::new(ProcedureStore::from_object_store(object_store.clone()));
-        let mut runner = new_runner(meta.clone(), Box::new(poison), procedure_store);
-        runner.manager_ctx.start();
-        runner.exponential_builder = ExponentialBuilder::default()
-            .with_min_delay(Duration::from_millis(1))
-            .with_max_times(3);
-        // Use the manager ctx as the context provider.
-        let ctx = context_with_provider(
-            meta.id,
-            runner.manager_ctx.clone() as Arc<dyn ContextProvider>,
-        );
-        // Manually add this procedure to the manager ctx.
-        runner
-            .manager_ctx
-            .procedures
-            .write()
-            .unwrap()
-            .insert(meta.id, runner.meta.clone());
-        // Run the runner and execute the procedure.
-        runner.execute_once_with_retry(&ctx).await;
-        let err = meta.state().error().unwrap().clone();
-        assert_matches!(&*err, Error::RetryTimesExceeded { .. });
-
-        // Check the poison is deleted.
-        let procedure_id = runner
-            .manager_ctx
-            .poison_manager
-            .get_poison(&poison_key.to_string())
-            .await
-            .unwrap();
-        assert_eq!(procedure_id, None);
-    }
-
     #[tokio::test]
     async fn test_execute_poisoned() {
         let mut times = 0;

src/common/query/src/error.rs

@@ -103,6 +103,13 @@ pub enum Error {
         source: common_recordbatch::error::Error,
     },
 
+    #[snafu(display("Failed to convert arrow schema"))]
+    ConvertArrowSchema {
+        #[snafu(implicit)]
+        location: Location,
+        source: DataTypeError,
+    },
+
     #[snafu(display("Failed to cast array to {:?}", typ))]
     TypeCast {
         #[snafu(source)]
@@ -237,6 +244,7 @@ impl ErrorExt for Error {
             Error::InvalidInputType { source, .. }
             | Error::IntoVector { source, .. }
             | Error::FromScalarValue { source, .. }
+            | Error::ConvertArrowSchema { source, .. }
             | Error::FromArrowArray { source, .. }
             | Error::InvalidVectorString { source, .. } => source.status_code(),
 

src/common/test-util/Cargo.toml

@@ -9,7 +9,6 @@ workspace = true
 
 [dependencies]
 client = { workspace = true, features = ["testing"] }
-common-grpc.workspace = true
 common-query.workspace = true
 common-recordbatch.workspace = true
 once_cell.workspace = true

src/common/test-util/src/flight.rs

@@ -1,26 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use common_grpc::flight::{FlightEncoder, FlightMessage};
-use common_grpc::FlightData;
-use common_recordbatch::DfRecordBatch;
-
-/// Encodes record batch to a Schema message and a RecordBatch message.
-pub fn encode_to_flight_data(rb: DfRecordBatch) -> (FlightData, FlightData) {
-    let mut encoder = FlightEncoder::default();
-    (
-        encoder.encode(FlightMessage::Schema(rb.schema())),
-        encoder.encode(FlightMessage::RecordBatch(rb)),
-    )
-}

src/common/test-util/src/lib.rs

@@ -16,7 +16,6 @@ use std::path::{Path, PathBuf};
 use std::process::Command;
 use std::sync::LazyLock;
 
-pub mod flight;
 pub mod ports;
 pub mod recordbatch;
 pub mod temp_dir;

src/flow/Cargo.toml

@@ -61,12 +61,10 @@ prost.workspace = true
 query.workspace = true
 rand.workspace = true
 serde.workspace = true
-serde_json.workspace = true
 servers.workspace = true
 session.workspace = true
 smallvec.workspace = true
 snafu.workspace = true
-sql.workspace = true
 store-api.workspace = true
 strum.workspace = true
 substrait.workspace = true

src/flow/src/adapter/flownode_impl.rs

@@ -18,7 +18,7 @@ use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
 
 use api::v1::flow::{
-    flow_request, AdjustFlow, CreateRequest, DropRequest, FlowRequest, FlowResponse, FlushFlow,
+    flow_request, CreateRequest, DropRequest, FlowRequest, FlowResponse, FlushFlow,
 };
 use api::v1::region::InsertRequests;
 use catalog::CatalogManager;
@@ -32,7 +32,6 @@ use common_telemetry::{error, info, trace, warn};
 use datatypes::value::Value;
 use futures::TryStreamExt;
 use itertools::Itertools;
-use serde::{Deserialize, Serialize};
 use session::context::QueryContextBuilder;
 use snafu::{ensure, IntoError, OptionExt, ResultExt};
 use store_api::storage::{RegionId, TableId};
@@ -360,7 +359,7 @@ impl FlowDualEngine {
                 }
             } else {
                 warn!(
-                    "Flows do not exist in flownode for node {:?}, flow_ids={:?}",
+                    "Flownode {:?} found flows not exist in flownode, flow_ids={:?}",
                     nodeid, to_be_created
                 );
             }
@@ -380,7 +379,7 @@ impl FlowDualEngine {
                 }
             } else {
                 warn!(
-                    "Flows do not exist in metadata for node {:?}, flow_ids={:?}",
+                    "Flownode {:?} found flows not exist in flownode, flow_ids={:?}",
                     nodeid, to_be_dropped
                 );
             }
@@ -810,25 +809,6 @@ impl common_meta::node_manager::Flownode for FlowDualEngine {
                     ..Default::default()
                 })
             }
-            Some(flow_request::Body::Adjust(AdjustFlow { flow_id, options })) => {
-                #[derive(Debug, Serialize, Deserialize)]
-                struct Options {
-                    min_run_interval_secs: u64,
-                    max_filter_num_per_query: usize,
-                }
-                let options: Options = serde_json::from_str(&options).with_context(|_| {
-                    common_meta::error::DeserializeFromJsonSnafu { input: options }
-                })?;
-                self.batching_engine
-                    .adjust_flow(
-                        flow_id.unwrap().id as u64,
-                        options.min_run_interval_secs,
-                        options.max_filter_num_per_query,
-                    )
-                    .await
-                    .map_err(to_meta_err(snafu::location!()))?;
-                Ok(Default::default())
-            }
             other => common_meta::error::InvalidFlowRequestBodySnafu { body: other }.fail(),
         }
     }
@@ -846,21 +826,100 @@ 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 {
-        match err {
-            crate::error::Error::FlowNotFound { id, .. } => {
-                common_meta::error::Error::FlowNotFound {
-                    flow_name: format!("flow_id={id}"),
-                    location,
-                }
-            }
-            _ => common_meta::error::Error::External {
-                location,
-                source: BoxedError::new(err),
-            },
+        common_meta::error::Error::External {
+            location,
+            source: BoxedError::new(err),
         }
     }
 }
 
+#[async_trait::async_trait]
+impl common_meta::node_manager::Flownode for StreamingEngine {
+    async fn handle(&self, request: FlowRequest) -> MetaResult<FlowResponse> {
+        let query_ctx = request
+            .header
+            .and_then(|h| h.query_context)
+            .map(|ctx| ctx.into());
+        match request.body {
+            Some(flow_request::Body::Create(CreateRequest {
+                flow_id: Some(task_id),
+                source_table_ids,
+                sink_table_name: Some(sink_table_name),
+                create_if_not_exists,
+                expire_after,
+                comment,
+                sql,
+                flow_options,
+                or_replace,
+            })) => {
+                let source_table_ids = source_table_ids.into_iter().map(|id| id.id).collect_vec();
+                let sink_table_name = [
+                    sink_table_name.catalog_name,
+                    sink_table_name.schema_name,
+                    sink_table_name.table_name,
+                ];
+                let expire_after = expire_after.map(|e| e.value);
+                let args = CreateFlowArgs {
+                    flow_id: task_id.id as u64,
+                    sink_table_name,
+                    source_table_ids,
+                    create_if_not_exists,
+                    or_replace,
+                    expire_after,
+                    comment: Some(comment),
+                    sql: sql.clone(),
+                    flow_options,
+                    query_ctx,
+                };
+                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
+                        .map(|id| greptime_proto::v1::FlowId { id: id as u32 })
+                        .into_iter()
+                        .collect_vec(),
+                    ..Default::default()
+                })
+            }
+            Some(flow_request::Body::Drop(DropRequest {
+                flow_id: Some(flow_id),
+            })) => {
+                self.remove_flow(flow_id.id as u64)
+                    .await
+                    .map_err(to_meta_err(snafu::location!()))?;
+                METRIC_FLOW_TASK_COUNT.dec();
+                Ok(Default::default())
+            }
+            Some(flow_request::Body::Flush(FlushFlow {
+                flow_id: Some(flow_id),
+            })) => {
+                let row = self
+                    .flush_flow_inner(flow_id.id as u64)
+                    .await
+                    .map_err(to_meta_err(snafu::location!()))?;
+                Ok(FlowResponse {
+                    affected_flows: vec![flow_id],
+                    affected_rows: row as u64,
+                    ..Default::default()
+                })
+            }
+            other => common_meta::error::InvalidFlowRequestBodySnafu { body: other }.fail(),
+        }
+    }
+
+    async fn handle_inserts(&self, request: InsertRequests) -> MetaResult<FlowResponse> {
+        self.handle_inserts_inner(request)
+            .await
+            .map(|_| Default::default())
+            .map_err(to_meta_err(snafu::location!()))
+    }
+}
+
 impl FlowEngine for StreamingEngine {
     async fn create_flow(&self, args: CreateFlowArgs) -> Result<Option<FlowId>, Error> {
         self.create_flow_inner(args).await

src/flow/src/batching_mode/engine.rs

@@ -39,8 +39,7 @@ use crate::batching_mode::time_window::{find_time_window_expr, TimeWindowExpr};
 use crate::batching_mode::utils::sql_to_df_plan;
 use crate::engine::FlowEngine;
 use crate::error::{
-    ExternalSnafu, FlowAlreadyExistSnafu, FlowNotFoundSnafu, TableNotFoundMetaSnafu,
-    UnexpectedSnafu, UnsupportedSnafu,
+    ExternalSnafu, FlowAlreadyExistSnafu, TableNotFoundMetaSnafu, UnexpectedSnafu, UnsupportedSnafu,
 };
 use crate::{CreateFlowArgs, Error, FlowId, TableName};
 
@@ -313,7 +312,7 @@ impl BatchingEngine {
                 .unwrap_or("None".to_string())
         );
 
-        let task = BatchingTask::try_new(
+        let task = BatchingTask::new(
             flow_id,
             &sql,
             plan,
@@ -324,7 +323,7 @@ impl BatchingEngine {
             query_ctx,
             self.catalog_manager.clone(),
             rx,
-        )?;
+        );
 
         let task_inner = task.clone();
         let engine = self.query_engine.clone();
@@ -350,8 +349,7 @@ impl BatchingEngine {
 
     pub async fn remove_flow_inner(&self, flow_id: FlowId) -> Result<(), Error> {
         if self.tasks.write().await.remove(&flow_id).is_none() {
-            warn!("Flow {flow_id} not found in tasks");
-            FlowNotFoundSnafu { id: flow_id }.fail()?;
+            warn!("Flow {flow_id} not found in tasks")
         }
         let Some(tx) = self.shutdown_txs.write().await.remove(&flow_id) else {
             UnexpectedSnafu {
@@ -368,7 +366,9 @@ impl BatchingEngine {
     pub async fn flush_flow_inner(&self, flow_id: FlowId) -> Result<usize, Error> {
         debug!("Try flush flow {flow_id}");
         let task = self.tasks.read().await.get(&flow_id).cloned();
-        let task = task.with_context(|| FlowNotFoundSnafu { id: flow_id })?;
+        let task = task.with_context(|| UnexpectedSnafu {
+            reason: format!("Can't found task for flow {flow_id}"),
+        })?;
 
         task.mark_all_windows_as_dirty()?;
 
@@ -388,20 +388,6 @@ impl BatchingEngine {
     pub async fn flow_exist_inner(&self, flow_id: FlowId) -> bool {
         self.tasks.read().await.contains_key(&flow_id)
     }
-
-    pub async fn adjust_flow(
-        &self,
-        flow_id: FlowId,
-        min_run_interval_secs: u64,
-        max_filter_num_per_query: usize,
-    ) -> Result<(), Error> {
-        let task = self.tasks.read().await.get(&flow_id).cloned();
-        let task = task.with_context(|| FlowNotFoundSnafu { id: flow_id })?;
-        debug!("Adjusting flow {flow_id} with min_run_interval_secs={} and max_filter_num_per_query={}", min_run_interval_secs, max_filter_num_per_query);
-        task.adjust(min_run_interval_secs, max_filter_num_per_query);
-
-        Ok(())
-    }
 }
 
 impl FlowEngine for BatchingEngine {

src/flow/src/batching_mode/state.rs

@@ -30,10 +30,6 @@ use crate::batching_mode::task::BatchingTask;
 use crate::batching_mode::time_window::TimeWindowExpr;
 use crate::batching_mode::MIN_REFRESH_DURATION;
 use crate::error::{DatatypesSnafu, InternalSnafu, TimeSnafu, UnexpectedSnafu};
-use crate::metrics::{
-    METRIC_FLOW_BATCHING_ENGINE_QUERY_TIME_RANGE, METRIC_FLOW_BATCHING_ENGINE_QUERY_WINDOW_CNT,
-    METRIC_FLOW_BATCHING_ENGINE_STALLED_QUERY_WINDOW_CNT,
-};
 use crate::{Error, FlowId};
 
 /// The state of the [`BatchingTask`].
@@ -53,11 +49,6 @@ pub struct TaskState {
     pub(crate) shutdown_rx: oneshot::Receiver<()>,
     /// Task handle
     pub(crate) task_handle: Option<tokio::task::JoinHandle<()>>,
-
-    /// min run interval in seconds
-    pub(crate) min_run_interval: Option<u64>,
-    /// max filter number per query
-    pub(crate) max_filter_num: Option<usize>,
 }
 impl TaskState {
     pub fn new(query_ctx: QueryContextRef, shutdown_rx: oneshot::Receiver<()>) -> Self {
@@ -69,8 +60,6 @@ impl TaskState {
             exec_state: ExecState::Idle,
             shutdown_rx,
             task_handle: None,
-            min_run_interval: None,
-            max_filter_num: None,
         }
     }
 
@@ -82,30 +71,18 @@ impl TaskState {
         self.last_update_time = Instant::now();
     }
 
-    /// Compute the next query delay based on the time window size or the last query duration.
-    /// Aiming to avoid too frequent queries. But also not too long delay.
-    /// The delay is computed as follows:
-    /// - If `time_window_size` is set, the delay is half the time window size, constrained to be
-    ///   at least `last_query_duration` and at most `max_timeout`.
-    /// - If `time_window_size` is not set, the delay defaults to `last_query_duration`, constrained
-    ///   to be at least `MIN_REFRESH_DURATION` and at most `max_timeout`.
+    /// wait for at least `last_query_duration`, at most `max_timeout` to start next query
     ///
-    /// If there are dirty time windows, the function returns an immediate execution time to clean them.
-    /// TODO: Make this behavior configurable.
+    /// if have more dirty time window, exec next query immediately
     pub fn get_next_start_query_time(
         &self,
         flow_id: FlowId,
-        _time_window_size: &Option<Duration>,
         max_timeout: Option<Duration>,
     ) -> Instant {
         let next_duration = max_timeout
             .unwrap_or(self.last_query_duration)
-            .min(self.last_query_duration)
-            .max(
-                self.min_run_interval
-                    .map(|s| Duration::from_secs(s))
-                    .unwrap_or(MIN_REFRESH_DURATION),
-            );
+            .min(self.last_query_duration);
+        let next_duration = next_duration.max(MIN_REFRESH_DURATION);
 
         // if have dirty time window, execute immediately to clean dirty time window
         if self.dirty_time_windows.windows.is_empty() {
@@ -135,10 +112,10 @@ impl DirtyTimeWindows {
     /// Time window merge distance
     ///
     /// TODO(discord9): make those configurable
-    pub const MERGE_DIST: i32 = 3;
+    const MERGE_DIST: i32 = 3;
 
     /// Maximum number of filters allowed in a single query
-    pub const MAX_FILTER_NUM: usize = 20;
+    const MAX_FILTER_NUM: usize = 20;
 
     /// Add lower bounds to the dirty time windows. Upper bounds are ignored.
     ///
@@ -162,16 +139,11 @@ impl DirtyTimeWindows {
     }
 
     /// Generate all filter expressions consuming all time windows
-    ///
-    /// there is two limits:
-    /// - shouldn't return a too long time range(<=`window_size * window_cnt`), so that the query can be executed in a reasonable time
-    /// - shouldn't return too many time range exprs, so that the query can be parsed properly instead of causing parser to overflow
     pub fn gen_filter_exprs(
         &mut self,
         col_name: &str,
         expire_lower_bound: Option<Timestamp>,
         window_size: chrono::Duration,
-        window_cnt: usize,
         flow_id: FlowId,
         task_ctx: Option<&BatchingTask>,
     ) -> Result<Option<datafusion_expr::Expr>, Error> {
@@ -209,33 +181,12 @@ impl DirtyTimeWindows {
             }
         }
 
-        // get the first `window_cnt` time windows
-        let max_time_range = window_size * window_cnt as i32;
-        let nth = {
-            let mut cur_time_range = chrono::Duration::zero();
-            let mut nth_key = None;
-            for (idx, (start, end)) in self.windows.iter().enumerate() {
-                // if time range is too long, stop
-                if cur_time_range > max_time_range {
-                    nth_key = Some(*start);
-                    break;
-                }
-
-                // if we have enough time windows, stop
-                if idx >= window_cnt {
-                    nth_key = Some(*start);
-                    break;
-                }
-
-                if let Some(end) = end {
-                    if let Some(x) = end.sub(start) {
-                        cur_time_range += x;
-                    }
-                }
-            }
-
-            nth_key
-        };
+        // get the first `MAX_FILTER_NUM` time windows
+        let nth = self
+            .windows
+            .iter()
+            .nth(Self::MAX_FILTER_NUM)
+            .map(|(key, _)| *key);
         let first_nth = {
             if let Some(nth) = nth {
                 let mut after = self.windows.split_off(&nth);
@@ -247,28 +198,6 @@ impl DirtyTimeWindows {
             }
         };
 
-        METRIC_FLOW_BATCHING_ENGINE_QUERY_WINDOW_CNT
-            .with_label_values(&[flow_id.to_string().as_str()])
-            .observe(first_nth.len() as f64);
-
-        METRIC_FLOW_BATCHING_ENGINE_STALLED_QUERY_WINDOW_CNT
-            .with_label_values(&[flow_id.to_string().as_str()])
-            .observe(self.windows.len() as f64);
-
-        let full_time_range = first_nth
-            .iter()
-            .fold(chrono::Duration::zero(), |acc, (start, end)| {
-                if let Some(end) = end {
-                    acc + end.sub(start).unwrap_or(chrono::Duration::zero())
-                } else {
-                    acc
-                }
-            })
-            .num_seconds() as f64;
-        METRIC_FLOW_BATCHING_ENGINE_QUERY_TIME_RANGE
-            .with_label_values(&[flow_id.to_string().as_str()])
-            .observe(full_time_range);
-
         let mut expr_lst = vec![];
         for (start, end) in first_nth.into_iter() {
             // align using time window exprs
@@ -330,8 +259,6 @@ impl DirtyTimeWindows {
     }
 
     /// Merge time windows that overlaps or get too close
-    ///
-    /// TODO(discord9): not merge and prefer to send smaller time windows? how?
     pub fn merge_dirty_time_windows(
         &mut self,
         window_size: chrono::Duration,
@@ -530,14 +457,7 @@ mod test {
                 .unwrap();
             assert_eq!(expected, dirty.windows);
             let filter_expr = dirty
-                .gen_filter_exprs(
-                    "ts",
-                    expire_lower_bound,
-                    window_size,
-                    DirtyTimeWindows::MAX_FILTER_NUM,
-                    0,
-                    None,
-                )
+                .gen_filter_exprs("ts", expire_lower_bound, window_size, 0, None)
                 .unwrap();
 
             let unparser = datafusion::sql::unparser::Unparser::default();

src/flow/src/batching_mode/task.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::collections::{BTreeSet, HashSet};
+use std::ops::Deref;
 use std::sync::{Arc, RwLock};
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
@@ -28,7 +29,6 @@ use datafusion::optimizer::analyzer::count_wildcard_rule::CountWildcardRule;
 use datafusion::optimizer::AnalyzerRule;
 use datafusion::sql::unparser::expr_to_sql;
 use datafusion_common::tree_node::{Transformed, TreeNode};
-use datafusion_common::DFSchemaRef;
 use datafusion_expr::{DmlStatement, LogicalPlan, WriteOp};
 use datatypes::prelude::ConcreteDataType;
 use datatypes::schema::{ColumnSchema, Schema};
@@ -37,8 +37,6 @@ use query::query_engine::DefaultSerializer;
 use query::QueryEngineRef;
 use session::context::QueryContextRef;
 use snafu::{ensure, OptionExt, ResultExt};
-use sql::parser::{ParseOptions, ParserContext};
-use sql::statements::statement::Statement;
 use substrait::{DFLogicalSubstraitConvertor, SubstraitPlan};
 use tokio::sync::oneshot;
 use tokio::sync::oneshot::error::TryRecvError;
@@ -46,7 +44,7 @@ use tokio::time::Instant;
 
 use crate::adapter::{AUTO_CREATED_PLACEHOLDER_TS_COL, AUTO_CREATED_UPDATE_AT_TS_COL};
 use crate::batching_mode::frontend_client::FrontendClient;
-use crate::batching_mode::state::{DirtyTimeWindows, TaskState};
+use crate::batching_mode::state::TaskState;
 use crate::batching_mode::time_window::TimeWindowExpr;
 use crate::batching_mode::utils::{
     get_table_info_df_schema, sql_to_df_plan, AddAutoColumnRewriter, AddFilterRewriter,
@@ -70,42 +68,13 @@ use crate::{Error, FlowId};
 pub struct TaskConfig {
     pub flow_id: FlowId,
     pub query: String,
-    /// output schema of the query
-    pub output_schema: DFSchemaRef,
+    plan: Arc<LogicalPlan>,
     pub time_window_expr: Option<TimeWindowExpr>,
     /// in seconds
     pub expire_after: Option<i64>,
     sink_table_name: [String; 3],
     pub source_table_names: HashSet<[String; 3]>,
     catalog_manager: CatalogManagerRef,
-    query_type: QueryType,
-}
-
-fn determine_query_type(query: &str, query_ctx: &QueryContextRef) -> Result<QueryType, Error> {
-    let stmts =
-        ParserContext::create_with_dialect(query, query_ctx.sql_dialect(), ParseOptions::default())
-            .map_err(BoxedError::new)
-            .context(ExternalSnafu)?;
-
-    ensure!(
-        stmts.len() == 1,
-        InvalidQuerySnafu {
-            reason: format!("Expect only one statement, found {}", stmts.len())
-        }
-    );
-    let stmt = &stmts[0];
-    match stmt {
-        Statement::Tql(_) => Ok(QueryType::Tql),
-        _ => Ok(QueryType::Sql),
-    }
-}
-
-#[derive(Debug, Clone)]
-enum QueryType {
-    /// query is a tql query
-    Tql,
-    /// query is a sql query
-    Sql,
 }
 
 #[derive(Clone)]
@@ -116,7 +85,7 @@ pub struct BatchingTask {
 
 impl BatchingTask {
     #[allow(clippy::too_many_arguments)]
-    pub fn try_new(
+    pub fn new(
         flow_id: FlowId,
         query: &str,
         plan: LogicalPlan,
@@ -127,27 +96,20 @@ impl BatchingTask {
         query_ctx: QueryContextRef,
         catalog_manager: CatalogManagerRef,
         shutdown_rx: oneshot::Receiver<()>,
-    ) -> Result<Self, Error> {
-        Ok(Self {
+    ) -> Self {
+        Self {
             config: Arc::new(TaskConfig {
                 flow_id,
                 query: query.to_string(),
+                plan: Arc::new(plan),
                 time_window_expr,
                 expire_after,
                 sink_table_name,
                 source_table_names: source_table_names.into_iter().collect(),
                 catalog_manager,
-                output_schema: plan.schema().clone(),
-                query_type: determine_query_type(query, &query_ctx)?,
             }),
             state: Arc::new(RwLock::new(TaskState::new(query_ctx, shutdown_rx))),
-        })
-    }
-
-    pub fn adjust(&self, min_run_interval_secs: u64, max_filter_num_per_query: usize) {
-        let mut state = self.state.write().unwrap();
-        state.min_run_interval = Some(min_run_interval_secs);
-        state.max_filter_num = Some(max_filter_num_per_query);
+        }
     }
 
     /// mark time window range (now - expire_after, now) as dirty (or (0, now) if expire_after not set)
@@ -393,6 +355,7 @@ impl BatchingTask {
             METRIC_FLOW_BATCHING_ENGINE_SLOW_QUERY
                 .with_label_values(&[
                     flow_id.to_string().as_str(),
+                    &plan.to_string(),
                     &peer_desc.unwrap_or_default().to_string(),
                 ])
                 .observe(elapsed.as_secs_f64());
@@ -417,52 +380,33 @@ impl BatchingTask {
         frontend_client: Arc<FrontendClient>,
     ) {
         loop {
-            // first check if shutdown signal is received
-            // if so, break the loop
-            {
-                let mut state = self.state.write().unwrap();
-                match state.shutdown_rx.try_recv() {
-                    Ok(()) => break,
-                    Err(TryRecvError::Closed) => {
-                        warn!(
-                            "Unexpected shutdown flow {}, shutdown anyway",
-                            self.config.flow_id
-                        );
-                        break;
-                    }
-                    Err(TryRecvError::Empty) => (),
-                }
-            }
-
-            let new_query = match self.gen_insert_plan(&engine).await {
-                Ok(new_query) => new_query,
-                Err(err) => {
-                    common_telemetry::error!(err; "Failed to generate query for flow={}", self.config.flow_id);
-                    // also sleep for a little while before try again to prevent flooding logs
-                    tokio::time::sleep(MIN_REFRESH_DURATION).await;
-                    continue;
+            let mut new_query = None;
+            let mut gen_and_exec = async || {
+                new_query = self.gen_insert_plan(&engine).await?;
+                if let Some(new_query) = &new_query {
+                    self.execute_logical_plan(&frontend_client, new_query).await
+                } else {
+                    Ok(None)
                 }
             };
-
-            let res = if let Some(new_query) = &new_query {
-                self.execute_logical_plan(&frontend_client, new_query).await
-            } else {
-                Ok(None)
-            };
-
-            match res {
+            match gen_and_exec().await {
                 // normal execute, sleep for some time before doing next query
                 Ok(Some(_)) => {
                     let sleep_until = {
-                        let state = self.state.write().unwrap();
-
+                        let mut state = self.state.write().unwrap();
+                        match state.shutdown_rx.try_recv() {
+                            Ok(()) => break,
+                            Err(TryRecvError::Closed) => {
+                                warn!(
+                                    "Unexpected shutdown flow {}, shutdown anyway",
+                                    self.config.flow_id
+                                );
+                                break;
+                            }
+                            Err(TryRecvError::Empty) => (),
+                        }
                         state.get_next_start_query_time(
                             self.config.flow_id,
-                            &self
-                                .config
-                                .time_window_expr
-                                .as_ref()
-                                .and_then(|t| *t.time_window_size()),
                             Some(DEFAULT_BATCHING_ENGINE_QUERY_TIMEOUT),
                         )
                     };
@@ -528,7 +472,7 @@ impl BatchingTask {
             .unwrap_or(u64::MIN);
 
         let low_bound = Timestamp::new_second(low_bound as i64);
-        let schema_len = self.config.output_schema.fields().len();
+        let schema_len = self.config.plan.schema().fields().len();
 
         let expire_time_window_bound = self
             .config
@@ -537,103 +481,104 @@ impl BatchingTask {
             .map(|expr| expr.eval(low_bound))
             .transpose()?;
 
-        let (Some((Some(l), Some(u))), QueryType::Sql) =
-            (expire_time_window_bound, &self.config.query_type)
-        else {
-            // either no time window or not a sql query, then just use the original query
-            // use sink_table_meta to add to query the `update_at` and `__ts_placeholder` column's value too for compatibility reason
-            debug!(
-                "Flow id = {:?}, can't get window size: precise_lower_bound={expire_time_window_bound:?}, using the same query", self.config.flow_id
-            );
-            // clean dirty time window too, this could be from create flow's check_execute
-            self.state.write().unwrap().dirty_time_windows.clean();
+        let new_plan = {
+            let expr = {
+                match expire_time_window_bound {
+                    Some((Some(l), Some(u))) => {
+                        let window_size = u.sub(&l).with_context(|| UnexpectedSnafu {
+                            reason: format!("Can't get window size from {u:?} - {l:?}"),
+                        })?;
+                        let col_name = self
+                            .config
+                            .time_window_expr
+                            .as_ref()
+                            .map(|expr| expr.column_name.clone())
+                            .with_context(|| UnexpectedSnafu {
+                                reason: format!(
+                                    "Flow id={:?}, Failed to get column name from time window expr",
+                                    self.config.flow_id
+                                ),
+                            })?;
 
-            // TODO(discord9): not add auto column for tql query?
+                        self.state
+                            .write()
+                            .unwrap()
+                            .dirty_time_windows
+                            .gen_filter_exprs(
+                                &col_name,
+                                Some(l),
+                                window_size,
+                                self.config.flow_id,
+                                Some(self),
+                            )?
+                    }
+                    _ => {
+                        // use sink_table_meta to add to query the `update_at` and `__ts_placeholder` column's value too for compatibility reason
+                        debug!(
+                            "Flow id = {:?}, can't get window size: precise_lower_bound={expire_time_window_bound:?}, using the same query", self.config.flow_id
+                        );
+                        // clean dirty time window too, this could be from create flow's check_execute
+                        self.state.write().unwrap().dirty_time_windows.clean();
+
+                        let mut add_auto_column =
+                            AddAutoColumnRewriter::new(sink_table_schema.clone());
+                        let plan = self
+                            .config
+                            .plan
+                            .deref()
+                            .clone()
+                            .rewrite(&mut add_auto_column)
+                            .with_context(|_| DatafusionSnafu {
+                                context: format!(
+                                    "Failed to rewrite plan:\n {}\n",
+                                    self.config.plan
+                                ),
+                            })?
+                            .data;
+                        let schema_len = plan.schema().fields().len();
+
+                        // since no time window lower/upper bound is found, just return the original query(with auto columns)
+                        return Ok(Some((plan, schema_len)));
+                    }
+                }
+            };
+
+            debug!(
+                "Flow id={:?}, Generated filter expr: {:?}",
+                self.config.flow_id,
+                expr.as_ref()
+                    .map(|expr| expr_to_sql(expr).with_context(|_| DatafusionSnafu {
+                        context: format!("Failed to generate filter expr from {expr:?}"),
+                    }))
+                    .transpose()?
+                    .map(|s| s.to_string())
+            );
+
+            let Some(expr) = expr else {
+                // no new data, hence no need to update
+                debug!("Flow id={:?}, no new data, not update", self.config.flow_id);
+                return Ok(None);
+            };
+
+            // TODO(discord9): add auto column or not? This might break compatibility for auto created sink table before this, but that's ok right?
+
+            let mut add_filter = AddFilterRewriter::new(expr);
             let mut add_auto_column = AddAutoColumnRewriter::new(sink_table_schema.clone());
 
             let plan = sql_to_df_plan(query_ctx.clone(), engine.clone(), &self.config.query, false)
                 .await?;
-
-            let plan = plan
+            let rewrite = plan
                 .clone()
-                .rewrite(&mut add_auto_column)
+                .rewrite(&mut add_filter)
+                .and_then(|p| p.data.rewrite(&mut add_auto_column))
                 .with_context(|_| DatafusionSnafu {
                     context: format!("Failed to rewrite plan:\n {}\n", plan),
                 })?
                 .data;
-            let schema_len = plan.schema().fields().len();
-
-            // since no time window lower/upper bound is found, just return the original query(with auto columns)
-            return Ok(Some((plan, schema_len)));
+            // only apply optimize after complex rewrite is done
+            apply_df_optimizer(rewrite).await?
         };
 
-        debug!(
-            "Flow id = {:?}, found time window: precise_lower_bound={:?}, precise_upper_bound={:?}",
-            self.config.flow_id, l, u
-        );
-        let window_size = u.sub(&l).with_context(|| UnexpectedSnafu {
-            reason: format!("Can't get window size from {u:?} - {l:?}"),
-        })?;
-        let col_name = self
-            .config
-            .time_window_expr
-            .as_ref()
-            .map(|expr| expr.column_name.clone())
-            .with_context(|| UnexpectedSnafu {
-                reason: format!(
-                    "Flow id={:?}, Failed to get column name from time window expr",
-                    self.config.flow_id
-                ),
-            })?;
-
-        let expr = {
-            let mut state = self.state.write().unwrap();
-            let max_window_cnt = state
-                .max_filter_num
-                .unwrap_or(DirtyTimeWindows::MAX_FILTER_NUM);
-            state.dirty_time_windows.gen_filter_exprs(
-                &col_name,
-                Some(l),
-                window_size,
-                max_window_cnt,
-                self.config.flow_id,
-                Some(self),
-            )?
-        };
-
-        debug!(
-            "Flow id={:?}, Generated filter expr: {:?}",
-            self.config.flow_id,
-            expr.as_ref()
-                .map(|expr| expr_to_sql(expr).with_context(|_| DatafusionSnafu {
-                    context: format!("Failed to generate filter expr from {expr:?}"),
-                }))
-                .transpose()?
-                .map(|s| s.to_string())
-        );
-
-        let Some(expr) = expr else {
-            // no new data, hence no need to update
-            debug!("Flow id={:?}, no new data, not update", self.config.flow_id);
-            return Ok(None);
-        };
-
-        let mut add_filter = AddFilterRewriter::new(expr);
-        let mut add_auto_column = AddAutoColumnRewriter::new(sink_table_schema.clone());
-
-        let plan =
-            sql_to_df_plan(query_ctx.clone(), engine.clone(), &self.config.query, false).await?;
-        let rewrite = plan
-            .clone()
-            .rewrite(&mut add_filter)
-            .and_then(|p| p.data.rewrite(&mut add_auto_column))
-            .with_context(|_| DatafusionSnafu {
-                context: format!("Failed to rewrite plan:\n {}\n", plan),
-            })?
-            .data;
-        // only apply optimize after complex rewrite is done
-        let new_plan = apply_df_optimizer(rewrite).await?;
-
         Ok(Some((new_plan, schema_len)))
     }
 }

src/flow/src/batching_mode/time_window.rs

@@ -55,9 +55,6 @@ use crate::error::{
 use crate::expr::error::DataTypeSnafu;
 use crate::Error;
 
-/// Represents a test timestamp in seconds since the Unix epoch.
-const DEFAULT_TEST_TIMESTAMP: Timestamp = Timestamp::new_second(17_0000_0000);
-
 /// Time window expr like `date_bin(INTERVAL '1' MINUTE, ts)`, this type help with
 /// evaluating the expr using given timestamp
 ///
@@ -73,7 +70,6 @@ pub struct TimeWindowExpr {
     pub column_name: String,
     logical_expr: Expr,
     df_schema: DFSchema,
-    eval_time_window_size: Option<std::time::Duration>,
 }
 
 impl std::fmt::Display for TimeWindowExpr {
@@ -88,11 +84,6 @@ impl std::fmt::Display for TimeWindowExpr {
 }
 
 impl TimeWindowExpr {
-    /// The time window size of the expr, get from calling `eval` with a test timestamp
-    pub fn time_window_size(&self) -> &Option<std::time::Duration> {
-        &self.eval_time_window_size
-    }
-
     pub fn from_expr(
         expr: &Expr,
         column_name: &str,
@@ -100,28 +91,12 @@ impl TimeWindowExpr {
         session: &SessionState,
     ) -> Result<Self, Error> {
         let phy_expr: PhysicalExprRef = to_phy_expr(expr, df_schema, session)?;
-        let mut zelf = Self {
+        Ok(Self {
             phy_expr,
             column_name: column_name.to_string(),
             logical_expr: expr.clone(),
             df_schema: df_schema.clone(),
-            eval_time_window_size: None,
-        };
-        let test_ts = DEFAULT_TEST_TIMESTAMP;
-        let (l, u) = zelf.eval(test_ts)?;
-        let time_window_size = match (l, u) {
-            (Some(l), Some(u)) => u.sub(&l).map(|r| r.to_std()).transpose().map_err(|_| {
-                UnexpectedSnafu {
-                    reason: format!(
-                        "Expect upper bound older than lower bound, found upper={u:?} and lower={l:?}"
-                    ),
-                }
-                .build()
-            })?,
-            _ => None,
-        };
-        zelf.eval_time_window_size = time_window_size;
-        Ok(zelf)
+        })
     }
 
     pub fn eval(