fix: plan disorder from upgrading datafusion (#6787)

* fix: plan disorder from upgrading datafusion

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

* update sqlness

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

* update sqlness again

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

---------

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

.github/scripts/check-version.sh

@@ -35,8 +35,8 @@ HIGHER_VERSION=$(printf "%s\n%s" "$CLEAN_CURRENT" "$CLEAN_LATEST" | sort -V | ta
 
 if [ "$HIGHER_VERSION" = "$CLEAN_CURRENT" ]; then
   echo "Current version ($CLEAN_CURRENT) is NEWER than or EQUAL to latest ($CLEAN_LATEST)"
-  echo "should-push-latest-tag=true" >> $GITHUB_OUTPUT
+  echo "is-current-version-latest=true" >> $GITHUB_OUTPUT
 else
   echo "Current version ($CLEAN_CURRENT) is OLDER than latest ($CLEAN_LATEST)"
-  echo "should-push-latest-tag=false" >> $GITHUB_OUTPUT
+  echo "is-current-version-latest=false" >> $GITHUB_OUTPUT
 fi

.github/scripts/update-dev-builder-version.sh

@@ -21,7 +21,7 @@ update_dev_builder_version() {
 
   # Commit the changes.
   git add Makefile
-  git commit -m "ci: update dev-builder image tag"
+  git commit -s -m "ci: update dev-builder image tag"
   git push origin $BRANCH_NAME
 
   # Create a Pull Request.

.github/workflows/develop.yml

@@ -12,6 +12,7 @@ on:
       - 'docker/**'
       - '.gitignore'
       - 'grafana/**'
+      - 'Makefile'
   workflow_dispatch:
 
 name: CI
@@ -709,7 +710,7 @@ jobs:
       - name: Install toolchain
         uses: actions-rust-lang/setup-rust-toolchain@v1
         with:
-            cache: false
+          cache: false
       - name: Rust Cache
         uses: Swatinem/rust-cache@v2
         with:

.github/workflows/docs.yml

@@ -10,6 +10,7 @@ on:
       - 'docker/**'
       - '.gitignore'
       - 'grafana/**'
+      - 'Makefile'
   push:
     branches:
       - main
@@ -21,6 +22,7 @@ on:
       - 'docker/**'
       - '.gitignore'
       - 'grafana/**'
+      - 'Makefile'
   workflow_dispatch:
 
 name: CI

.github/workflows/release.yml

@@ -111,7 +111,8 @@ jobs:
       # The 'version' use as the global tag name of the release workflow.
       version: ${{ steps.create-version.outputs.version }}
 
-      should-push-latest-tag: ${{ steps.check-version.outputs.should-push-latest-tag }}
+      # The 'is-current-version-latest' determines whether to update 'latest' Docker tags and downstream repositories.
+      is-current-version-latest: ${{ steps.check-version.outputs.is-current-version-latest }}
     steps:
       - name: Checkout
         uses: actions/checkout@v4
@@ -321,7 +322,7 @@ jobs:
           image-registry-username: ${{ secrets.DOCKERHUB_USERNAME }}
           image-registry-password: ${{ secrets.DOCKERHUB_TOKEN }}
           version: ${{ needs.allocate-runners.outputs.version }}
-          push-latest-tag: ${{ needs.allocate-runners.outputs.should-push-latest-tag == 'true' && github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
+          push-latest-tag: ${{ needs.allocate-runners.outputs.is-current-version-latest == 'true' && github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
 
       - name: Set build image result
         id: set-build-image-result
@@ -368,7 +369,7 @@ jobs:
           dev-mode: false
           upload-to-s3: true
           update-version-info: true
-          push-latest-tag: ${{ needs.allocate-runners.outputs.should-push-latest-tag == 'true' && github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
+          push-latest-tag: ${{ needs.allocate-runners.outputs.is-current-version-latest == 'true' && github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
 
   publish-github-release:
     name: Create GitHub release and upload artifacts
@@ -476,7 +477,7 @@ jobs:
 
   bump-helm-charts-version:
     name: Bump helm charts version
-    if: ${{ github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
+    if: ${{ github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' && needs.allocate-runners.outputs.is-current-version-latest == 'true' }}
     needs: [allocate-runners, publish-github-release]
     runs-on: ubuntu-latest
     permissions:
@@ -497,7 +498,7 @@ jobs:
 
   bump-homebrew-greptime-version:
     name: Bump homebrew greptime version
-    if: ${{ github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
+    if: ${{ github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' && needs.allocate-runners.outputs.is-current-version-latest == 'true' }}
     needs: [allocate-runners, publish-github-release]
     runs-on: ubuntu-latest
     permissions:

.gitignore

@@ -60,4 +60,7 @@ tests-fuzz/corpus/
 greptimedb_data
 
 # github
-!/.github
+!/.github
+
+# Claude code
+CLAUDE.md

AUTHOR.md

@@ -10,12 +10,10 @@
 * [NiwakaDev](https://github.com/NiwakaDev)
 * [tisonkun](https://github.com/tisonkun)
 
-
 ## Team Members (in alphabetical order)
 
 * [apdong2022](https://github.com/apdong2022)
 * [beryl678](https://github.com/beryl678)
-* [Breeze-P](https://github.com/Breeze-P)
 * [daviderli614](https://github.com/daviderli614)
 * [discord9](https://github.com/discord9)
 * [evenyag](https://github.com/evenyag)

CONTRIBUTING.md

@@ -55,8 +55,9 @@ GreptimeDB uses the [Apache 2.0 license](https://github.com/GreptimeTeam/greptim
 - To ensure that community is free and confident in its ability to use your contributions, please sign the Contributor License Agreement (CLA) which will be incorporated in the pull request process.
 - Make sure all files have proper license header (running `docker run --rm -v $(pwd):/github/workspace ghcr.io/korandoru/hawkeye-native:v3 format` from the project root).
 - Make sure all your codes are formatted and follow the [coding style](https://pingcap.github.io/style-guide/rust/) and [style guide](docs/style-guide.md).
-- Make sure all unit tests are passed using [nextest](https://nexte.st/index.html) `cargo nextest run`.
-- Make sure all clippy warnings are fixed (you can check it locally by running `cargo clippy --workspace --all-targets -- -D warnings`).
+- Make sure all unit tests are passed using [nextest](https://nexte.st/index.html) `cargo nextest run --workspace --features pg_kvbackend,mysql_kvbackend` or `make test`.
+- Make sure all clippy warnings are fixed (you can check it locally by running `cargo clippy --workspace --all-targets -- -D warnings` or `make clippy`).
+- When modifying sample configuration files in `config/`, run `make config-docs` (which requires Docker to be installed) to update the configuration documentation and include it in your commit.
 
 #### `pre-commit` Hooks
 

Cargo.toml

@@ -13,6 +13,7 @@ members = [
     "src/common/datasource",
     "src/common/decimal",
     "src/common/error",
+    "src/common/event-recorder",
     "src/common/frontend",
     "src/common/function",
     "src/common/greptimedb-telemetry",
@@ -72,7 +73,7 @@ members = [
 resolver = "2"
 
 [workspace.package]
-version = "0.16.0"
+version = "0.17.0"
 edition = "2021"
 license = "Apache-2.0"
 
@@ -97,11 +98,12 @@ rust.unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tokio_unstable)'] }
 # See for more detaiils: https://github.com/rust-lang/cargo/issues/11329
 ahash = { version = "0.8", features = ["compile-time-rng"] }
 aquamarine = "0.6"
-arrow = { version = "54.2", features = ["prettyprint"] }
-arrow-array = { version = "54.2", default-features = false, features = ["chrono-tz"] }
-arrow-flight = "54.2"
-arrow-ipc = { version = "54.2", default-features = false, features = ["lz4", "zstd"] }
-arrow-schema = { version = "54.2", features = ["serde"] }
+arrow = { version = "56.0", features = ["prettyprint"] }
+arrow-array = { version = "56.0", default-features = false, features = ["chrono-tz"] }
+arrow-buffer = "56.0"
+arrow-flight = "56.0"
+arrow-ipc = { version = "56.0", default-features = false, features = ["lz4", "zstd"] }
+arrow-schema = { version = "56.0", features = ["serde"] }
 async-stream = "0.3"
 async-trait = "0.1"
 # Remember to update axum-extra, axum-macros when updating axum
@@ -120,26 +122,27 @@ clap = { version = "4.4", features = ["derive"] }
 config = "0.13.0"
 crossbeam-utils = "0.8"
 dashmap = "6.1"
-datafusion = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
-datafusion-common = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
-datafusion-expr = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
-datafusion-functions = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
-datafusion-functions-aggregate-common = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
-datafusion-optimizer = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
-datafusion-physical-expr = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
-datafusion-physical-plan = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
-datafusion-sql = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
-datafusion-substrait = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
+datafusion = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7d5214512740b4dfb742b6b3d91ed9affcc2c9d0" }
+datafusion-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7d5214512740b4dfb742b6b3d91ed9affcc2c9d0" }
+datafusion-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7d5214512740b4dfb742b6b3d91ed9affcc2c9d0" }
+datafusion-functions = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7d5214512740b4dfb742b6b3d91ed9affcc2c9d0" }
+datafusion-functions-aggregate-common = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7d5214512740b4dfb742b6b3d91ed9affcc2c9d0" }
+datafusion-optimizer = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7d5214512740b4dfb742b6b3d91ed9affcc2c9d0" }
+datafusion-orc = { git = "https://github.com/GreptimeTeam/datafusion-orc", rev = "a0a5f902158f153119316eaeec868cff3fc8a99d" }
+datafusion-physical-expr = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7d5214512740b4dfb742b6b3d91ed9affcc2c9d0" }
+datafusion-physical-plan = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7d5214512740b4dfb742b6b3d91ed9affcc2c9d0" }
+datafusion-sql = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7d5214512740b4dfb742b6b3d91ed9affcc2c9d0" }
+datafusion-substrait = { git = "https://github.com/GreptimeTeam/datafusion.git", rev = "7d5214512740b4dfb742b6b3d91ed9affcc2c9d0" }
 deadpool = "0.12"
 deadpool-postgres = "0.14"
 derive_builder = "0.20"
 dotenv = "0.15"
 either = "1.15"
-etcd-client = "0.14"
+etcd-client = { git = "https://github.com/GreptimeTeam/etcd-client", rev = "f62df834f0cffda355eba96691fe1a9a332b75a7" }
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "ceb1af4fa9309ce65bda0367db7b384df2bb4d4f" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "df2bb74b5990c159dfd5b7a344eecf8f4307af64" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"
@@ -150,7 +153,7 @@ itertools = "0.14"
 jsonb = { git = "https://github.com/databendlabs/jsonb.git", rev = "8c8d2fc294a39f3ff08909d60f718639cfba3875", default-features = false }
 lazy_static = "1.4"
 local-ip-address = "0.6"
-loki-proto = { git = "https://github.com/GreptimeTeam/loki-proto.git", rev = "1434ecf23a2654025d86188fb5205e7a74b225d3" }
+loki-proto = { git = "https://github.com/GreptimeTeam/loki-proto.git", rev = "3b7cd33234358b18ece977bf689dc6fb760f29ab" }
 meter-core = { git = "https://github.com/GreptimeTeam/greptime-meter.git", rev = "5618e779cf2bb4755b499c630fba4c35e91898cb" }
 mockall = "0.13"
 moka = "0.12"
@@ -158,22 +161,25 @@ nalgebra = "0.33"
 nix = { version = "0.30.1", default-features = false, features = ["event", "fs", "process"] }
 notify = "8.0"
 num_cpus = "1.16"
-object_store_opendal = "0.50"
+object_store_opendal = "0.54"
 once_cell = "1.18"
-opentelemetry-proto = { version = "0.27", features = [
+opentelemetry-proto = { version = "0.30", features = [
     "gen-tonic",
     "metrics",
     "trace",
     "with-serde",
     "logs",
 ] }
+ordered-float = { version = "4.3", features = ["serde"] }
 parking_lot = "0.12"
-parquet = { version = "54.2", default-features = false, features = ["arrow", "async", "object_store"] }
+parquet = { version = "56.0", default-features = false, features = ["arrow", "async", "object_store"] }
 paste = "1.0"
 pin-project = "1.0"
+pretty_assertions = "1.4.0"
 prometheus = { version = "0.13.3", features = ["process"] }
 promql-parser = { version = "0.6", features = ["ser"] }
 prost = { version = "0.13", features = ["no-recursion-limit"] }
+prost-types = "0.13"
 raft-engine = { version = "0.4.1", default-features = false }
 rand = "0.9"
 ratelimit = "0.10"
@@ -185,7 +191,7 @@ reqwest = { version = "0.12", default-features = false, features = [
     "stream",
     "multipart",
 ] }
-rskafka = { git = "https://github.com/influxdata/rskafka.git", rev = "8dbd01ed809f5a791833a594e85b144e36e45820", features = [
+rskafka = { git = "https://github.com/influxdata/rskafka.git", rev = "a62120b6c74d68953464b256f858dc1c41a903b4", features = [
     "transport-tls",
 ] }
 rstest = "0.25"
@@ -194,18 +200,18 @@ rust_decimal = "1.33"
 rustc-hash = "2.0"
 # It is worth noting that we should try to avoid using aws-lc-rs until it can be compiled on various platforms.
 rustls = { version = "0.23.25", default-features = false }
+sea-query = "0.32"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1.0", features = ["float_roundtrip"] }
 serde_with = "3"
-shadow-rs = "1.1"
 simd-json = "0.15"
 similar-asserts = "1.6.0"
 smallvec = { version = "1", features = ["serde"] }
 snafu = "0.8"
-sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "0cf6c04490d59435ee965edd2078e8855bd8471e", features = [
+sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "39e4fc94c3c741981f77e9d63b5ce8c02e0a27ea", features = [
     "visitor",
     "serde",
-] } # branch = "v0.54.x"
+] } # branch = "v0.55.x"
 sqlx = { version = "0.8", features = [
     "runtime-tokio-rustls",
     "mysql",
@@ -215,18 +221,21 @@ sqlx = { version = "0.8", features = [
 strum = { version = "0.27", features = ["derive"] }
 sysinfo = "0.33"
 tempfile = "3"
-tokio = { version = "1.40", features = ["full"] }
+tokio = { version = "1.47", features = ["full"] }
 tokio-postgres = "0.7"
 tokio-rustls = { version = "0.26.2", default-features = false }
 tokio-stream = "0.1"
 tokio-util = { version = "0.7", features = ["io-util", "compat"] }
 toml = "0.8.8"
-tonic = { version = "0.12", features = ["tls", "gzip", "zstd"] }
+tonic = { version = "0.13", features = ["tls-ring", "gzip", "zstd"] }
 tower = "0.5"
+tower-http = "0.6"
+tracing = "0.1"
 tracing-appender = "0.2"
 tracing-subscriber = { version = "0.3", features = ["env-filter", "json", "fmt"] }
 typetag = "0.2"
-uuid = { version = "1.7", features = ["serde", "v4", "fast-rng"] }
+uuid = { version = "1.17", features = ["serde", "v4", "fast-rng"] }
+vrl = "0.25"
 zstd = "0.13"
 # DO_NOT_REMOVE_THIS: END_OF_EXTERNAL_DEPENDENCIES
 
@@ -244,6 +253,7 @@ common-config = { path = "src/common/config" }
 common-datasource = { path = "src/common/datasource" }
 common-decimal = { path = "src/common/decimal" }
 common-error = { path = "src/common/error" }
+common-event-recorder = { path = "src/common/event-recorder" }
 common-frontend = { path = "src/common/frontend" }
 common-function = { path = "src/common/function" }
 common-greptimedb-telemetry = { path = "src/common/greptimedb-telemetry" }
@@ -283,7 +293,7 @@ mito-codec = { path = "src/mito-codec" }
 mito2 = { path = "src/mito2" }
 object-store = { path = "src/object-store" }
 operator = { path = "src/operator" }
-otel-arrow-rust = { git = "https://github.com/open-telemetry/otel-arrow", rev = "5d551412d2a12e689cde4d84c14ef29e36784e51", features = [
+otel-arrow-rust = { git = "https://github.com/GreptimeTeam/otel-arrow", rev = "2d64b7c0fa95642028a8205b36fe9ea0b023ec59", features = [
     "server",
 ] }
 partition = { path = "src/partition" }

Makefile

@@ -8,7 +8,7 @@ CARGO_BUILD_OPTS := --locked
 IMAGE_REGISTRY ?= docker.io
 IMAGE_NAMESPACE ?= greptime
 IMAGE_TAG ?= latest
-DEV_BUILDER_IMAGE_TAG ?= 2025-05-19-b2377d4b-20250520045554
+DEV_BUILDER_IMAGE_TAG ?= 2025-05-19-32619816-20250818043248
 BUILDX_MULTI_PLATFORM_BUILD ?= false
 BUILDX_BUILDER_NAME ?= gtbuilder
 BASE_IMAGE ?= ubuntu
@@ -22,7 +22,7 @@ SQLNESS_OPTS ?=
 ETCD_VERSION ?= v3.5.9
 ETCD_IMAGE ?= quay.io/coreos/etcd:${ETCD_VERSION}
 RETRY_COUNT ?= 3
-NEXTEST_OPTS := --retries ${RETRY_COUNT}
+NEXTEST_OPTS := --retries ${RETRY_COUNT} --features pg_kvbackend,mysql_kvbackend
 BUILD_JOBS ?= $(shell which nproc 1>/dev/null && expr $$(nproc) / 2) # If nproc is not available, we don't set the build jobs.
 ifeq ($(BUILD_JOBS), 0) # If the number of cores is less than 2, set the build jobs to 1.
   BUILD_JOBS := 1

config/config.md

@@ -41,6 +41,7 @@
 | `mysql.addr` | String | `127.0.0.1:4002` | The addr to bind the MySQL server. |
 | `mysql.runtime_size` | Integer | `2` | The number of server worker threads. |
 | `mysql.keep_alive` | String | `0s` | Server-side keep-alive time.<br/>Set to 0 (default) to disable. |
+| `mysql.prepared_stmt_cache_size` | Integer | `10000` | Maximum entries in the MySQL prepared statement cache; default is 10,000. |
 | `mysql.tls` | -- | -- | -- |
 | `mysql.tls.mode` | String | `disable` | TLS mode, refer to https://www.postgresql.org/docs/current/libpq-ssl.html<br/>- `disable` (default value)<br/>- `prefer`<br/>- `require`<br/>- `verify-ca`<br/>- `verify-full` |
 | `mysql.tls.cert_path` | String | Unset | Certificate file path. |
@@ -147,6 +148,7 @@
 | `region_engine.mito.write_cache_ttl` | String | Unset | TTL for write cache. |
 | `region_engine.mito.sst_write_buffer_size` | String | `8MB` | Buffer size for SST writing. |
 | `region_engine.mito.parallel_scan_channel_size` | Integer | `32` | Capacity of the channel to send data from parallel scan tasks to the main task. |
+| `region_engine.mito.max_concurrent_scan_files` | Integer | `128` | Maximum number of SST files to scan concurrently. |
 | `region_engine.mito.allow_stale_entries` | Bool | `false` | Whether to allow stale WAL entries read during replay. |
 | `region_engine.mito.min_compaction_interval` | String | `0m` | Minimum time interval between two compactions.<br/>To align with the old behavior, the default value is 0 (no restrictions). |
 | `region_engine.mito.index` | -- | -- | The options for index in Mito engine. |
@@ -185,12 +187,13 @@
 | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. |
 | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. |
 | `logging.enable_otlp_tracing` | Bool | `false` | Enable OTLP tracing. |
-| `logging.otlp_endpoint` | String | `http://localhost:4318` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4318/v1/traces` | The OTLP tracing endpoint. |
 | `logging.append_stdout` | Bool | `true` | Whether to append logs to stdout. |
 | `logging.log_format` | String | `text` | The log format. Can be `text`/`json`. |
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
 | `logging.otlp_export_protocol` | String | `http` | The OTLP tracing export protocol. Can be `grpc`/`http`. |
-| `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
+| `logging.otlp_headers` | -- | -- | Additional OTLP headers, only valid when using OTLP http |
+| `logging.tracing_sample_ratio` | -- | Unset | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
 | `slow_query` | -- | -- | The slow query log options. |
 | `slow_query.enable` | Bool | `false` | Whether to enable slow query log. |
@@ -207,6 +210,8 @@
 | `export_metrics.remote_write.headers` | InlineTable | -- | HTTP headers of Prometheus remote-write carry. |
 | `tracing` | -- | -- | The tracing options. Only effect when compiled with `tokio-console` feature. |
 | `tracing.tokio_console_addr` | String | Unset | The tokio console address. |
+| `memory` | -- | -- | The memory options. |
+| `memory.enable_heap_profiling` | Bool | `true` | Whether to enable heap profiling activation during startup.<br/>When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable<br/>is set to "prof:true,prof_active:false". The official image adds this env variable.<br/>Default is true. |
 
 
 ## Distributed Mode
@@ -245,6 +250,7 @@
 | `mysql.addr` | String | `127.0.0.1:4002` | The addr to bind the MySQL server. |
 | `mysql.runtime_size` | Integer | `2` | The number of server worker threads. |
 | `mysql.keep_alive` | String | `0s` | Server-side keep-alive time.<br/>Set to 0 (default) to disable. |
+| `mysql.prepared_stmt_cache_size` | Integer | `10000` | Maximum entries in the MySQL prepared statement cache; default is 10,000. |
 | `mysql.tls` | -- | -- | -- |
 | `mysql.tls.mode` | String | `disable` | TLS mode, refer to https://www.postgresql.org/docs/current/libpq-ssl.html<br/>- `disable` (default value)<br/>- `prefer`<br/>- `require`<br/>- `verify-ca`<br/>- `verify-full` |
 | `mysql.tls.cert_path` | String | Unset | Certificate file path. |
@@ -281,6 +287,7 @@
 | `meta_client.metadata_cache_tti` | String | `5m` | -- |
 | `query` | -- | -- | The query engine options. |
 | `query.parallelism` | Integer | `0` | Parallelism of the query engine.<br/>Default to 0, which means the number of CPU cores. |
+| `query.allow_query_fallback` | Bool | `false` | Whether to allow query fallback when push down optimize fails.<br/>Default to false, meaning when push down optimize failed, return error msg |
 | `datanode` | -- | -- | Datanode options. |
 | `datanode.client` | -- | -- | Datanode client options. |
 | `datanode.client.connect_timeout` | String | `10s` | -- |
@@ -289,19 +296,20 @@
 | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. |
 | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. |
 | `logging.enable_otlp_tracing` | Bool | `false` | Enable OTLP tracing. |
-| `logging.otlp_endpoint` | String | `http://localhost:4318` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4318/v1/traces` | The OTLP tracing endpoint. |
 | `logging.append_stdout` | Bool | `true` | Whether to append logs to stdout. |
 | `logging.log_format` | String | `text` | The log format. Can be `text`/`json`. |
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
 | `logging.otlp_export_protocol` | String | `http` | The OTLP tracing export protocol. Can be `grpc`/`http`. |
-| `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
+| `logging.otlp_headers` | -- | -- | Additional OTLP headers, only valid when using OTLP http |
+| `logging.tracing_sample_ratio` | -- | Unset | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
 | `slow_query` | -- | -- | The slow query log options. |
 | `slow_query.enable` | Bool | `true` | Whether to enable slow query log. |
 | `slow_query.record_type` | String | `system_table` | The record type of slow queries. It can be `system_table` or `log`.<br/>If `system_table` is selected, the slow queries will be recorded in a system table `greptime_private.slow_queries`.<br/>If `log` is selected, the slow queries will be logged in a log file `greptimedb-slow-queries.*`. |
 | `slow_query.threshold` | String | `30s` | The threshold of slow query. It can be human readable time string, for example: `10s`, `100ms`, `1s`. |
 | `slow_query.sample_ratio` | Float | `1.0` | The sampling ratio of slow query log. The value should be in the range of (0, 1]. For example, `0.1` means 10% of the slow queries will be logged and `1.0` means all slow queries will be logged. |
-| `slow_query.ttl` | String | `30d` | The TTL of the `slow_queries` system table. Default is `30d` when `record_type` is `system_table`. |
+| `slow_query.ttl` | String | `90d` | The TTL of the `slow_queries` system table. Default is `90d` when `record_type` is `system_table`. |
 | `export_metrics` | -- | -- | The frontend can export its metrics and send to Prometheus compatible service (e.g. `greptimedb` itself) from remote-write API.<br/>This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape. |
 | `export_metrics.enable` | Bool | `false` | whether enable export metrics. |
 | `export_metrics.write_interval` | String | `30s` | The interval of export metrics. |
@@ -310,6 +318,10 @@
 | `export_metrics.remote_write.headers` | InlineTable | -- | HTTP headers of Prometheus remote-write carry. |
 | `tracing` | -- | -- | The tracing options. Only effect when compiled with `tokio-console` feature. |
 | `tracing.tokio_console_addr` | String | Unset | The tokio console address. |
+| `memory` | -- | -- | The memory options. |
+| `memory.enable_heap_profiling` | Bool | `true` | Whether to enable heap profiling activation during startup.<br/>When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable<br/>is set to "prof:true,prof_active:false". The official image adds this env variable.<br/>Default is true. |
+| `event_recorder` | -- | -- | Configuration options for the event recorder. |
+| `event_recorder.ttl` | String | `90d` | TTL for the events table that will be used to store the events. Default is `90d`. |
 
 
 ### Metasrv
@@ -332,6 +344,12 @@
 | `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. |
+| `backend_tls` | -- | -- | TLS configuration for kv store backend (only applicable for PostgreSQL/MySQL backends)<br/>When using PostgreSQL or MySQL as metadata store, you can configure TLS here |
+| `backend_tls.mode` | String | `prefer` | TLS mode, refer to https://www.postgresql.org/docs/current/libpq-ssl.html<br/>- "disable" - No TLS<br/>- "prefer" (default) - Try TLS, fallback to plain<br/>- "require" - Require TLS<br/>- "verify_ca" - Require TLS and verify CA<br/>- "verify_full" - Require TLS and verify hostname |
+| `backend_tls.cert_path` | String | `""` | Path to client certificate file (for client authentication)<br/>Like "/path/to/client.crt" |
+| `backend_tls.key_path` | String | `""` | Path to client private key file (for client authentication)<br/>Like "/path/to/client.key" |
+| `backend_tls.ca_cert_path` | String | `""` | Path to CA certificate file (for server certificate verification)<br/>Required when using custom CAs or self-signed certificates<br/>Leave empty to use system root certificates only<br/>Like "/path/to/ca.crt" |
+| `backend_tls.watch` | Bool | `false` | Watch for certificate file changes and auto reload |
 | `grpc` | -- | -- | The gRPC server options. |
 | `grpc.bind_addr` | String | `127.0.0.1:3002` | The address to bind the gRPC server. |
 | `grpc.server_addr` | String | `127.0.0.1:3002` | The communication server address for the frontend and datanode to connect to metasrv.<br/>If left empty or unset, the server will automatically use the IP address of the first network interface<br/>on the host, with the same port number as the one specified in `bind_addr`. |
@@ -359,26 +377,32 @@
 | `datanode.client.tcp_nodelay` | Bool | `true` | `TCP_NODELAY` option for accepted connections. |
 | `wal` | -- | -- | -- |
 | `wal.provider` | String | `raft_engine` | -- |
-| `wal.broker_endpoints` | Array | -- | The broker endpoints of the Kafka cluster. |
-| `wal.auto_create_topics` | Bool | `true` | Automatically create topics for WAL.<br/>Set to `true` to automatically create topics for WAL.<br/>Otherwise, use topics named `topic_name_prefix_[0..num_topics)` |
-| `wal.auto_prune_interval` | String | `0s` | Interval of automatically WAL pruning.<br/>Set to `0s` to disable automatically WAL pruning which delete unused remote WAL entries periodically. |
-| `wal.trigger_flush_threshold` | Integer | `0` | The threshold to trigger a flush operation of a region in automatically WAL pruning.<br/>Metasrv will send a flush request to flush the region when:<br/>`trigger_flush_threshold` + `prunable_entry_id` < `max_prunable_entry_id`<br/>where:<br/>- `prunable_entry_id` is the maximum entry id that can be pruned of the region.<br/>- `max_prunable_entry_id` is the maximum prunable entry id among all regions in the same topic.<br/>Set to `0` to disable the flush operation. |
-| `wal.auto_prune_parallelism` | Integer | `10` | Concurrent task limit for automatically WAL pruning. |
-| `wal.num_topics` | Integer | `64` | Number of topics. |
-| `wal.selector_type` | String | `round_robin` | Topic selector type.<br/>Available selector types:<br/>- `round_robin` (default) |
-| `wal.topic_name_prefix` | String | `greptimedb_wal_topic` | A Kafka topic is constructed by concatenating `topic_name_prefix` and `topic_id`.<br/>Only accepts strings that match the following regular expression pattern:<br/>[a-zA-Z_:-][a-zA-Z0-9_:\-\.@#]*<br/>i.g., greptimedb_wal_topic_0, greptimedb_wal_topic_1. |
-| `wal.replication_factor` | Integer | `1` | Expected number of replicas of each partition. |
-| `wal.create_topic_timeout` | String | `30s` | Above which a topic creation operation will be cancelled. |
+| `wal.broker_endpoints` | Array | -- | The broker endpoints of the Kafka cluster.<br/><br/>**It's only used when the provider is `kafka`**. |
+| `wal.auto_create_topics` | Bool | `true` | Automatically create topics for WAL.<br/>Set to `true` to automatically create topics for WAL.<br/>Otherwise, use topics named `topic_name_prefix_[0..num_topics)`<br/>**It's only used when the provider is `kafka`**. |
+| `wal.auto_prune_interval` | String | `10m` | Interval of automatically WAL pruning.<br/>Set to `0s` to disable automatically WAL pruning which delete unused remote WAL entries periodically.<br/>**It's only used when the provider is `kafka`**. |
+| `wal.flush_trigger_size` | String | `512MB` | Estimated size threshold to trigger a flush when using Kafka remote WAL.<br/>Since multiple regions may share a Kafka topic, the estimated size is calculated as:<br/>  (latest_entry_id - flushed_entry_id) * avg_record_size<br/>MetaSrv triggers a flush for a region when this estimated size exceeds `flush_trigger_size`.<br/>- `latest_entry_id`: The latest entry ID in the topic.<br/>- `flushed_entry_id`: The last flushed entry ID for the region.<br/>Set to "0" to let the system decide the flush trigger size.<br/>**It's only used when the provider is `kafka`**. |
+| `wal.auto_prune_parallelism` | Integer | `10` | Concurrent task limit for automatically WAL pruning.<br/>**It's only used when the provider is `kafka`**. |
+| `wal.num_topics` | Integer | `64` | Number of topics used for remote WAL.<br/>**It's only used when the provider is `kafka`**. |
+| `wal.selector_type` | String | `round_robin` | Topic selector type.<br/>Available selector types:<br/>- `round_robin` (default)<br/>**It's only used when the provider is `kafka`**. |
+| `wal.topic_name_prefix` | String | `greptimedb_wal_topic` | A Kafka topic is constructed by concatenating `topic_name_prefix` and `topic_id`.<br/>Only accepts strings that match the following regular expression pattern:<br/>[a-zA-Z_:-][a-zA-Z0-9_:\-\.@#]*<br/>i.g., greptimedb_wal_topic_0, greptimedb_wal_topic_1.<br/>**It's only used when the provider is `kafka`**. |
+| `wal.replication_factor` | Integer | `1` | Expected number of replicas of each partition.<br/>**It's only used when the provider is `kafka`**. |
+| `wal.create_topic_timeout` | String | `30s` | The timeout for creating a Kafka topic.<br/>**It's only used when the provider is `kafka`**. |
+| `event_recorder` | -- | -- | Configuration options for the event recorder. |
+| `event_recorder.ttl` | String | `90d` | TTL for the events table that will be used to store the events. Default is `90d`. |
+| `stats_persistence` | -- | -- | Configuration options for the stats persistence. |
+| `stats_persistence.ttl` | String | `30d` | TTL for the stats table that will be used to store the stats. Default is `30d`.<br/>Set to `0s` to disable stats persistence. |
+| `stats_persistence.interval` | String | `60s` | The interval to persist the stats. Default is `60s`.<br/>The minimum value is `60s`, if the value is less than `60s`, it will be overridden to `60s`. |
 | `logging` | -- | -- | The logging options. |
 | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. |
 | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. |
 | `logging.enable_otlp_tracing` | Bool | `false` | Enable OTLP tracing. |
-| `logging.otlp_endpoint` | String | `http://localhost:4318` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4318/v1/traces` | The OTLP tracing endpoint. |
 | `logging.append_stdout` | Bool | `true` | Whether to append logs to stdout. |
 | `logging.log_format` | String | `text` | The log format. Can be `text`/`json`. |
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
 | `logging.otlp_export_protocol` | String | `http` | The OTLP tracing export protocol. Can be `grpc`/`http`. |
-| `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
+| `logging.otlp_headers` | -- | -- | Additional OTLP headers, only valid when using OTLP http |
+| `logging.tracing_sample_ratio` | -- | Unset | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
 | `export_metrics` | -- | -- | The metasrv can export its metrics and send to Prometheus compatible service (e.g. `greptimedb` itself) from remote-write API.<br/>This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape. |
 | `export_metrics.enable` | Bool | `false` | whether enable export metrics. |
@@ -388,6 +412,8 @@
 | `export_metrics.remote_write.headers` | InlineTable | -- | HTTP headers of Prometheus remote-write carry. |
 | `tracing` | -- | -- | The tracing options. Only effect when compiled with `tokio-console` feature. |
 | `tracing.tokio_console_addr` | String | Unset | The tokio console address. |
+| `memory` | -- | -- | The memory options. |
+| `memory.enable_heap_profiling` | Bool | `true` | Whether to enable heap profiling activation during startup.<br/>When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable<br/>is set to "prof:true,prof_active:false". The official image adds this env variable.<br/>Default is true. |
 
 
 ### Datanode
@@ -500,6 +526,7 @@
 | `region_engine.mito.write_cache_ttl` | String | Unset | TTL for write cache. |
 | `region_engine.mito.sst_write_buffer_size` | String | `8MB` | Buffer size for SST writing. |
 | `region_engine.mito.parallel_scan_channel_size` | Integer | `32` | Capacity of the channel to send data from parallel scan tasks to the main task. |
+| `region_engine.mito.max_concurrent_scan_files` | Integer | `128` | Maximum number of SST files to scan concurrently. |
 | `region_engine.mito.allow_stale_entries` | Bool | `false` | Whether to allow stale WAL entries read during replay. |
 | `region_engine.mito.min_compaction_interval` | String | `0m` | Minimum time interval between two compactions.<br/>To align with the old behavior, the default value is 0 (no restrictions). |
 | `region_engine.mito.index` | -- | -- | The options for index in Mito engine. |
@@ -538,12 +565,13 @@
 | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. |
 | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. |
 | `logging.enable_otlp_tracing` | Bool | `false` | Enable OTLP tracing. |
-| `logging.otlp_endpoint` | String | `http://localhost:4318` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4318/v1/traces` | The OTLP tracing endpoint. |
 | `logging.append_stdout` | Bool | `true` | Whether to append logs to stdout. |
 | `logging.log_format` | String | `text` | The log format. Can be `text`/`json`. |
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
 | `logging.otlp_export_protocol` | String | `http` | The OTLP tracing export protocol. Can be `grpc`/`http`. |
-| `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
+| `logging.otlp_headers` | -- | -- | Additional OTLP headers, only valid when using OTLP http |
+| `logging.tracing_sample_ratio` | -- | Unset | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
 | `export_metrics` | -- | -- | The datanode can export its metrics and send to Prometheus compatible service (e.g. `greptimedb` itself) from remote-write API.<br/>This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape. |
 | `export_metrics.enable` | Bool | `false` | whether enable export metrics. |
@@ -553,6 +581,8 @@
 | `export_metrics.remote_write.headers` | InlineTable | -- | HTTP headers of Prometheus remote-write carry. |
 | `tracing` | -- | -- | The tracing options. Only effect when compiled with `tokio-console` feature. |
 | `tracing.tokio_console_addr` | String | Unset | The tokio console address. |
+| `memory` | -- | -- | The memory options. |
+| `memory.enable_heap_profiling` | Bool | `true` | Whether to enable heap profiling activation during startup.<br/>When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable<br/>is set to "prof:true,prof_active:false". The official image adds this env variable.<br/>Default is true. |
 
 
 ### Flownode
@@ -562,6 +592,22 @@
 | `node_id` | Integer | Unset | The flownode identifier and should be unique in the cluster. |
 | `flow` | -- | -- | flow engine options. |
 | `flow.num_workers` | Integer | `0` | The number of flow worker in flownode.<br/>Not setting(or set to 0) this value will use the number of CPU cores divided by 2. |
+| `flow.batching_mode` | -- | -- | -- |
+| `flow.batching_mode.query_timeout` | String | `600s` | The default batching engine query timeout is 10 minutes. |
+| `flow.batching_mode.slow_query_threshold` | String | `60s` | will output a warn log for any query that runs for more that this threshold |
+| `flow.batching_mode.experimental_min_refresh_duration` | String | `5s` | The minimum duration between two queries execution by batching mode task |
+| `flow.batching_mode.grpc_conn_timeout` | String | `5s` | The gRPC connection timeout |
+| `flow.batching_mode.experimental_grpc_max_retries` | Integer | `3` | The gRPC max retry number |
+| `flow.batching_mode.experimental_frontend_scan_timeout` | String | `30s` | Flow wait for available frontend timeout,<br/>if failed to find available frontend after frontend_scan_timeout elapsed, return error<br/>which prevent flownode from starting |
+| `flow.batching_mode.experimental_frontend_activity_timeout` | String | `60s` | Frontend activity timeout<br/>if frontend is down(not sending heartbeat) for more than frontend_activity_timeout,<br/>it will be removed from the list that flownode use to connect |
+| `flow.batching_mode.experimental_max_filter_num_per_query` | Integer | `20` | Maximum number of filters allowed in a single query |
+| `flow.batching_mode.experimental_time_window_merge_threshold` | Integer | `3` | Time window merge distance |
+| `flow.batching_mode.read_preference` | String | `Leader` | Read preference of the Frontend client. |
+| `flow.batching_mode.frontend_tls` | -- | -- | -- |
+| `flow.batching_mode.frontend_tls.enabled` | Bool | `false` | Whether to enable TLS for client. |
+| `flow.batching_mode.frontend_tls.server_ca_cert_path` | String | Unset | Server Certificate file path. |
+| `flow.batching_mode.frontend_tls.client_cert_path` | String | Unset | Client Certificate file path. |
+| `flow.batching_mode.frontend_tls.client_key_path` | String | Unset | Client Private key file path. |
 | `grpc` | -- | -- | The gRPC server options. |
 | `grpc.bind_addr` | String | `127.0.0.1:6800` | The address to bind the gRPC server. |
 | `grpc.server_addr` | String | `127.0.0.1:6800` | The address advertised to the metasrv,<br/>and used for connections from outside the host |
@@ -589,14 +635,17 @@
 | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. |
 | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. |
 | `logging.enable_otlp_tracing` | Bool | `false` | Enable OTLP tracing. |
-| `logging.otlp_endpoint` | String | `http://localhost:4318` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4318/v1/traces` | The OTLP tracing endpoint. |
 | `logging.append_stdout` | Bool | `true` | Whether to append logs to stdout. |
 | `logging.log_format` | String | `text` | The log format. Can be `text`/`json`. |
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
 | `logging.otlp_export_protocol` | String | `http` | The OTLP tracing export protocol. Can be `grpc`/`http`. |
-| `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
+| `logging.otlp_headers` | -- | -- | Additional OTLP headers, only valid when using OTLP http |
+| `logging.tracing_sample_ratio` | -- | Unset | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
 | `tracing` | -- | -- | The tracing options. Only effect when compiled with `tokio-console` feature. |
 | `tracing.tokio_console_addr` | String | Unset | The tokio console address. |
 | `query` | -- | -- | -- |
 | `query.parallelism` | Integer | `1` | Parallelism of the query engine for query sent by flownode.<br/>Default to 1, so it won't use too much cpu or memory |
+| `memory` | -- | -- | The memory options. |
+| `memory.enable_heap_profiling` | Bool | `true` | Whether to enable heap profiling activation during startup.<br/>When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable<br/>is set to "prof:true,prof_active:false". The official image adds this env variable.<br/>Default is true. |

config/datanode.example.toml

@@ -474,6 +474,9 @@ sst_write_buffer_size = "8MB"
 ## Capacity of the channel to send data from parallel scan tasks to the main task.
 parallel_scan_channel_size = 32
 
+## Maximum number of SST files to scan concurrently.
+max_concurrent_scan_files = 128
+
 ## Whether to allow stale WAL entries read during replay.
 allow_stale_entries = false
 
@@ -629,7 +632,7 @@ level = "info"
 enable_otlp_tracing = false
 
 ## The OTLP tracing endpoint.
-otlp_endpoint = "http://localhost:4318"
+otlp_endpoint = "http://localhost:4318/v1/traces"
 
 ## Whether to append logs to stdout.
 append_stdout = true
@@ -643,6 +646,13 @@ max_log_files = 720
 ## The OTLP tracing export protocol. Can be `grpc`/`http`.
 otlp_export_protocol = "http"
 
+## Additional OTLP headers, only valid when using OTLP http
+[logging.otlp_headers]
+## @toml2docs:none-default
+#Authorization = "Bearer my-token"
+## @toml2docs:none-default
+#Database = "My database"
+
 ## The percentage of tracing will be sampled and exported.
 ## Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.
 ## ratio > 1 are treated as 1. Fractions < 0 are treated as 0
@@ -669,3 +679,11 @@ headers = { }
 ## The tokio console address.
 ## @toml2docs:none-default
 #+ tokio_console_addr = "127.0.0.1"
+
+## The memory options.
+[memory]
+## Whether to enable heap profiling activation during startup.
+## When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable
+## is set to "prof:true,prof_active:false". The official image adds this env variable.
+## Default is true.
+enable_heap_profiling = true

config/flownode.example.toml

@@ -7,6 +7,43 @@ node_id = 14
 ## The number of flow worker in flownode.
 ## Not setting(or set to 0) this value will use the number of CPU cores divided by 2.
 #+num_workers=0
+[flow.batching_mode]
+## The default batching engine query timeout is 10 minutes.
+#+query_timeout="600s"
+## will output a warn log for any query that runs for more that this threshold
+#+slow_query_threshold="60s"
+## The minimum duration between two queries execution by batching mode task
+#+experimental_min_refresh_duration="5s"
+## The gRPC connection timeout
+#+grpc_conn_timeout="5s"
+## The gRPC max retry number
+#+experimental_grpc_max_retries=3
+## Flow wait for available frontend timeout,
+## if failed to find available frontend after frontend_scan_timeout elapsed, return error
+## which prevent flownode from starting
+#+experimental_frontend_scan_timeout="30s"
+## Frontend activity timeout
+## if frontend is down(not sending heartbeat) for more than frontend_activity_timeout,
+## it will be removed from the list that flownode use to connect
+#+experimental_frontend_activity_timeout="60s"
+## Maximum number of filters allowed in a single query
+#+experimental_max_filter_num_per_query=20
+## Time window merge distance
+#+experimental_time_window_merge_threshold=3
+## Read preference of the Frontend client.
+#+read_preference="Leader"
+[flow.batching_mode.frontend_tls]
+## Whether to enable TLS for client.
+#+enabled=false
+## Server Certificate file path.
+## @toml2docs:none-default
+#+server_ca_cert_path=""
+## Client Certificate file path.
+## @toml2docs:none-default
+#+client_cert_path=""
+## Client Private key file path.
+## @toml2docs:none-default
+#+client_key_path=""
 
 ## The gRPC server options.
 [grpc]
@@ -83,7 +120,7 @@ level = "info"
 enable_otlp_tracing = false
 
 ## The OTLP tracing endpoint.
-otlp_endpoint = "http://localhost:4318"
+otlp_endpoint = "http://localhost:4318/v1/traces"
 
 ## Whether to append logs to stdout.
 append_stdout = true
@@ -97,6 +134,13 @@ max_log_files = 720
 ## The OTLP tracing export protocol. Can be `grpc`/`http`.
 otlp_export_protocol = "http"
 
+## Additional OTLP headers, only valid when using OTLP http
+[logging.otlp_headers]
+## @toml2docs:none-default
+#Authorization = "Bearer my-token"
+## @toml2docs:none-default
+#Database = "My database"
+
 ## The percentage of tracing will be sampled and exported.
 ## Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.
 ## ratio > 1 are treated as 1. Fractions < 0 are treated as 0
@@ -113,3 +157,11 @@ default_ratio = 1.0
 ## Parallelism of the query engine for query sent by flownode.
 ## Default to 1, so it won't use too much cpu or memory
 parallelism = 1
+
+## The memory options.
+[memory]
+## Whether to enable heap profiling activation during startup.
+## When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable
+## is set to "prof:true,prof_active:false". The official image adds this env variable.
+## Default is true.
+enable_heap_profiling = true

config/frontend.example.toml

@@ -90,6 +90,8 @@ runtime_size = 2
 ## Server-side keep-alive time.
 ## Set to 0 (default) to disable.
 keep_alive = "0s"
+## Maximum entries in the MySQL prepared statement cache; default is 10,000.
+prepared_stmt_cache_size = 10000
 
 # MySQL server TLS options.
 [mysql.tls]
@@ -197,6 +199,9 @@ metadata_cache_tti = "5m"
 ## Parallelism of the query engine.
 ## Default to 0, which means the number of CPU cores.
 parallelism = 0
+## Whether to allow query fallback when push down optimize fails.
+## Default to false, meaning when push down optimize failed, return error msg
+allow_query_fallback = false
 
 ## Datanode options.
 [datanode]
@@ -218,7 +223,7 @@ level = "info"
 enable_otlp_tracing = false
 
 ## The OTLP tracing endpoint.
-otlp_endpoint = "http://localhost:4318"
+otlp_endpoint = "http://localhost:4318/v1/traces"
 
 ## Whether to append logs to stdout.
 append_stdout = true
@@ -232,6 +237,13 @@ max_log_files = 720
 ## The OTLP tracing export protocol. Can be `grpc`/`http`.
 otlp_export_protocol = "http"
 
+## Additional OTLP headers, only valid when using OTLP http
+[logging.otlp_headers]
+## @toml2docs:none-default
+#Authorization = "Bearer my-token"
+## @toml2docs:none-default
+#Database = "My database"
+
 ## The percentage of tracing will be sampled and exported.
 ## Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.
 ## ratio > 1 are treated as 1. Fractions < 0 are treated as 0
@@ -254,8 +266,8 @@ threshold = "30s"
 ## The sampling ratio of slow query log. The value should be in the range of (0, 1]. For example, `0.1` means 10% of the slow queries will be logged and `1.0` means all slow queries will be logged.
 sample_ratio = 1.0
 
-## The TTL of the `slow_queries` system table. Default is `30d` when `record_type` is `system_table`.
-ttl = "30d"
+## The TTL of the `slow_queries` system table. Default is `90d` when `record_type` is `system_table`.
+ttl = "90d"
 
 ## The frontend can export its metrics and send to Prometheus compatible service (e.g. `greptimedb` itself) from remote-write API.
 ## This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape.
@@ -277,3 +289,16 @@ headers = { }
 ## The tokio console address.
 ## @toml2docs:none-default
 #+ tokio_console_addr = "127.0.0.1"
+
+## The memory options.
+[memory]
+## Whether to enable heap profiling activation during startup.
+## When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable
+## is set to "prof:true,prof_active:false". The official image adds this env variable.
+## Default is true.
+enable_heap_profiling = true
+
+## Configuration options for the event recorder.
+[event_recorder]
+## TTL for the events table that will be used to store the events. Default is `90d`.
+ttl = "90d"

config/metasrv.example.toml

@@ -65,6 +65,34 @@ node_max_idle_time = "24hours"
 ## The number of threads to execute the runtime for global write operations.
 #+ compact_rt_size = 4
 
+## TLS configuration for kv store backend (only applicable for PostgreSQL/MySQL backends)
+## When using PostgreSQL or MySQL as metadata store, you can configure TLS here
+[backend_tls]
+## TLS mode, refer to https://www.postgresql.org/docs/current/libpq-ssl.html
+## - "disable" - No TLS
+## - "prefer" (default) - Try TLS, fallback to plain
+## - "require" - Require TLS
+## - "verify_ca" - Require TLS and verify CA
+## - "verify_full" - Require TLS and verify hostname
+mode = "prefer"
+
+## Path to client certificate file (for client authentication)
+## Like "/path/to/client.crt"
+cert_path = ""
+
+## Path to client private key file (for client authentication)
+## Like "/path/to/client.key"
+key_path = ""
+
+## Path to CA certificate file (for server certificate verification)
+## Required when using custom CAs or self-signed certificates
+## Leave empty to use system root certificates only
+## Like "/path/to/ca.crt"
+ca_cert_path = ""
+
+## Watch for certificate file changes and auto reload
+watch = false
+
 ## The gRPC server options.
 [grpc]
 ## The address to bind the gRPC server.
@@ -148,50 +176,61 @@ tcp_nodelay = true
 # - `kafka`: metasrv **have to be** configured with kafka wal config when using kafka wal provider in datanode.
 provider = "raft_engine"
 
-# Kafka wal config.
-
 ## The broker endpoints of the Kafka cluster.
+##
+## **It's only used when the provider is `kafka`**.
 broker_endpoints = ["127.0.0.1:9092"]
 
 ## Automatically create topics for WAL.
 ## Set to `true` to automatically create topics for WAL.
 ## Otherwise, use topics named `topic_name_prefix_[0..num_topics)`
+## **It's only used when the provider is `kafka`**.
 auto_create_topics = true
 
 ## Interval of automatically WAL pruning.
 ## Set to `0s` to disable automatically WAL pruning which delete unused remote WAL entries periodically.
-auto_prune_interval = "0s"
+## **It's only used when the provider is `kafka`**.
+auto_prune_interval = "10m"
 
-## The threshold to trigger a flush operation of a region in automatically WAL pruning.
-## Metasrv will send a flush request to flush the region when:
-## `trigger_flush_threshold` + `prunable_entry_id` < `max_prunable_entry_id`
-## where:
-## - `prunable_entry_id` is the maximum entry id that can be pruned of the region.
-## - `max_prunable_entry_id` is the maximum prunable entry id among all regions in the same topic.
-## Set to `0` to disable the flush operation.
-trigger_flush_threshold = 0
+
+## Estimated size threshold to trigger a flush when using Kafka remote WAL.
+## Since multiple regions may share a Kafka topic, the estimated size is calculated as:
+##   (latest_entry_id - flushed_entry_id) * avg_record_size
+## MetaSrv triggers a flush for a region when this estimated size exceeds `flush_trigger_size`.
+## - `latest_entry_id`: The latest entry ID in the topic.
+## - `flushed_entry_id`: The last flushed entry ID for the region.
+## Set to "0" to let the system decide the flush trigger size.
+## **It's only used when the provider is `kafka`**.
+flush_trigger_size = "512MB"
 
 ## Concurrent task limit for automatically WAL pruning.
+## **It's only used when the provider is `kafka`**.
 auto_prune_parallelism = 10
 
-## Number of topics.
+## Number of topics used for remote WAL.
+## **It's only used when the provider is `kafka`**.
 num_topics = 64
 
 ## Topic selector type.
 ## Available selector types:
 ## - `round_robin` (default)
+## **It's only used when the provider is `kafka`**.
 selector_type = "round_robin"
 
+
 ## A Kafka topic is constructed by concatenating `topic_name_prefix` and `topic_id`.
 ## Only accepts strings that match the following regular expression pattern:
 ## [a-zA-Z_:-][a-zA-Z0-9_:\-\.@#]*
 ## i.g., greptimedb_wal_topic_0, greptimedb_wal_topic_1.
+## **It's only used when the provider is `kafka`**.
 topic_name_prefix = "greptimedb_wal_topic"
 
 ## Expected number of replicas of each partition.
+## **It's only used when the provider is `kafka`**.
 replication_factor = 1
 
-## Above which a topic creation operation will be cancelled.
+## The timeout for creating a Kafka topic.
+## **It's only used when the provider is `kafka`**.
 create_topic_timeout = "30s"
 
 # The Kafka SASL configuration.
@@ -212,6 +251,20 @@ create_topic_timeout = "30s"
 # client_cert_path = "/path/to/client_cert"
 # client_key_path = "/path/to/key"
 
+## Configuration options for the event recorder.
+[event_recorder]
+## TTL for the events table that will be used to store the events. Default is `90d`.
+ttl = "90d"
+
+## Configuration options for the stats persistence.
+[stats_persistence]
+## TTL for the stats table that will be used to store the stats. Default is `30d`.
+## Set to `0s` to disable stats persistence.
+ttl = "30d"
+## The interval to persist the stats. Default is `60s`.
+## The minimum value is `60s`, if the value is less than `60s`, it will be overridden to `60s`.
+interval = "60s"
+
 ## The logging options.
 [logging]
 ## The directory to store the log files. If set to empty, logs will not be written to files.
@@ -225,7 +278,7 @@ level = "info"
 enable_otlp_tracing = false
 
 ## The OTLP tracing endpoint.
-otlp_endpoint = "http://localhost:4318"
+otlp_endpoint = "http://localhost:4318/v1/traces"
 
 ## Whether to append logs to stdout.
 append_stdout = true
@@ -239,6 +292,14 @@ max_log_files = 720
 ## The OTLP tracing export protocol. Can be `grpc`/`http`.
 otlp_export_protocol = "http"
 
+## Additional OTLP headers, only valid when using OTLP http
+[logging.otlp_headers]
+## @toml2docs:none-default
+#Authorization = "Bearer my-token"
+## @toml2docs:none-default
+#Database = "My database"
+
+
 ## The percentage of tracing will be sampled and exported.
 ## Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.
 ## ratio > 1 are treated as 1. Fractions < 0 are treated as 0
@@ -265,3 +326,11 @@ headers = { }
 ## The tokio console address.
 ## @toml2docs:none-default
 #+ tokio_console_addr = "127.0.0.1"
+
+## The memory options.
+[memory]
+## Whether to enable heap profiling activation during startup.
+## When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable
+## is set to "prof:true,prof_active:false". The official image adds this env variable.
+## Default is true.
+enable_heap_profiling = true

config/standalone.example.toml

@@ -85,7 +85,8 @@ runtime_size = 2
 ## Server-side keep-alive time.
 ## Set to 0 (default) to disable.
 keep_alive = "0s"
-
+## Maximum entries in the MySQL prepared statement cache; default is 10,000.
+prepared_stmt_cache_size= 10000
 # MySQL server TLS options.
 [mysql.tls]
 
@@ -565,6 +566,9 @@ sst_write_buffer_size = "8MB"
 ## Capacity of the channel to send data from parallel scan tasks to the main task.
 parallel_scan_channel_size = 32
 
+## Maximum number of SST files to scan concurrently.
+max_concurrent_scan_files = 128
+
 ## Whether to allow stale WAL entries read during replay.
 allow_stale_entries = false
 
@@ -720,7 +724,7 @@ level = "info"
 enable_otlp_tracing = false
 
 ## The OTLP tracing endpoint.
-otlp_endpoint = "http://localhost:4318"
+otlp_endpoint = "http://localhost:4318/v1/traces"
 
 ## Whether to append logs to stdout.
 append_stdout = true
@@ -734,6 +738,13 @@ max_log_files = 720
 ## The OTLP tracing export protocol. Can be `grpc`/`http`.
 otlp_export_protocol = "http"
 
+## Additional OTLP headers, only valid when using OTLP http
+[logging.otlp_headers]
+## @toml2docs:none-default
+#Authorization = "Bearer my-token"
+## @toml2docs:none-default
+#Database = "My database"
+
 ## The percentage of tracing will be sampled and exported.
 ## Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.
 ## ratio > 1 are treated as 1. Fractions < 0 are treated as 0
@@ -783,3 +794,11 @@ headers = { }
 ## The tokio console address.
 ## @toml2docs:none-default
 #+ tokio_console_addr = "127.0.0.1"
+
+## The memory options.
+[memory]
+## Whether to enable heap profiling activation during startup.
+## When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable
+## is set to "prof:true,prof_active:false". The official image adds this env variable.
+## Default is true.
+enable_heap_profiling = true

docker/buildx/centos/Dockerfile

@@ -47,4 +47,6 @@ WORKDIR /greptime
 COPY --from=builder /out/target/${OUTPUT_DIR}/greptime /greptime/bin/
 ENV PATH /greptime/bin/:$PATH
 
+ENV MALLOC_CONF="prof:true,prof_active:false"
+
 ENTRYPOINT ["greptime"]

docker/buildx/ubuntu/Dockerfile

@@ -47,4 +47,6 @@ WORKDIR /greptime
 COPY --from=builder /out/target/${OUTPUT_DIR}/greptime /greptime/bin/
 ENV PATH /greptime/bin/:$PATH
 
+ENV MALLOC_CONF="prof:true,prof_active:false"
+
 ENTRYPOINT ["greptime"]

docker/ci/centos/Dockerfile

@@ -15,4 +15,6 @@ ADD $TARGETARCH/greptime /greptime/bin/
 
 ENV PATH /greptime/bin/:$PATH
 
+ENV MALLOC_CONF="prof:true,prof_active:false"
+
 ENTRYPOINT ["greptime"]

docker/ci/ubuntu/Dockerfile

@@ -18,4 +18,6 @@ ENV PATH /greptime/bin/:$PATH
 
 ENV TARGET_BIN=$TARGET_BIN
 
+ENV MALLOC_CONF="prof:true,prof_active:false"
+
 ENTRYPOINT ["sh", "-c", "exec $TARGET_BIN \"$@\"", "--"]

docker/dev-builder/centos/Dockerfile

@@ -19,7 +19,7 @@ ARG PROTOBUF_VERSION=29.3
 
 RUN curl -OL https://github.com/protocolbuffers/protobuf/releases/download/v${PROTOBUF_VERSION}/protoc-${PROTOBUF_VERSION}-linux-x86_64.zip && \
     unzip protoc-${PROTOBUF_VERSION}-linux-x86_64.zip -d protoc3;
-    
+
 RUN mv protoc3/bin/* /usr/local/bin/
 RUN mv protoc3/include/* /usr/local/include/
 

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

@@ -30,6 +30,23 @@ curl https://raw.githubusercontent.com/brendangregg/FlameGraph/master/flamegraph
 
 ## Profiling
 
+### Configuration
+
+You can control heap profiling activation through configuration. Add the following to your configuration file:
+
+```toml
+[memory]
+# Whether to enable heap profiling activation during startup.
+# When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable
+# is set to "prof:true,prof_active:false". The official image adds this env variable.
+# Default is true.
+enable_heap_profiling = true
+```
+
+By default, if you set `MALLOC_CONF=prof:true,prof_active:false`, the database will enable profiling during startup. You can disable this behavior by setting `enable_heap_profiling = false` in the configuration.
+
+### Starting with environment variables
+
 Start GreptimeDB instance with environment variables:
 
 ```bash
@@ -40,6 +57,23 @@ MALLOC_CONF=prof:true ./target/debug/greptime standalone start
 _RJEM_MALLOC_CONF=prof:true ./target/debug/greptime standalone start
 ```
 
+### Memory profiling control
+
+You can control heap profiling activation using the new HTTP APIs:
+
+```bash
+# Check current profiling status
+curl -X GET localhost:4000/debug/prof/mem/status
+
+# Activate heap profiling (if not already active)
+curl -X POST localhost:4000/debug/prof/mem/activate
+
+# Deactivate heap profiling
+curl -X POST localhost:4000/debug/prof/mem/deactivate
+```
+
+### Dump memory profiling data
+
 Dump memory profiling data through HTTP API:
 
 ```bash

docs/how-to/how-to-write-aggregate-function.md

@@ -1,72 +0,0 @@
-Currently, our query engine is based on DataFusion, so all aggregate function is executed by DataFusion, through its UDAF interface. You can find DataFusion's UDAF example [here](https://github.com/apache/arrow-datafusion/blob/arrow2/datafusion-examples/examples/simple_udaf.rs). Basically, we provide the same way as DataFusion to write aggregate functions: both are centered in a struct called "Accumulator" to accumulates states along the way in aggregation.
-
-However, DataFusion's UDAF implementation has a huge restriction, that it requires user to provide a concrete "Accumulator". Take `Median` aggregate function for example, to aggregate a `u32` datatype column, you have to write a `MedianU32`, and use `SELECT MEDIANU32(x)` in SQL. `MedianU32` cannot be used to aggregate a `i32` datatype column. Or, there's another way: you can use a special type that can hold all kinds of data (like our `Value` enum or Arrow's `ScalarValue`), and `match` all the way up to do aggregate calculations. It might work, though rather tedious. (But I think it's DataFusion's preferred way to write UDAF.)
-
-So is there a way we can make an aggregate function that automatically match the input data's type? For example, a `Median` aggregator that can work on both `u32` column and `i32`? The answer is yes until we find a way to bypass DataFusion's restriction, a restriction that DataFusion simply doesn't pass the input data's type when creating an Accumulator.
-
-> There's an example in `my_sum_udaf_example.rs`, take that as quick start.
-
-# 1. Impl `AggregateFunctionCreator` trait for your accumulator creator.
-
-You must first define a struct that will be used to create your accumulator. For example,
-
-```Rust
-#[as_aggr_func_creator]
-#[derive(Debug, AggrFuncTypeStore)]
-struct MySumAccumulatorCreator {}
-```
-
-Attribute macro `#[as_aggr_func_creator]` and derive macro `#[derive(Debug, AggrFuncTypeStore)]` must both be annotated on the struct. They work together to provide a storage of aggregate function's input data types, which are needed for creating generic accumulator later.  
-
-> Note that the `as_aggr_func_creator` macro will add fields to the struct, so the struct cannot be defined as an empty struct without field like `struct Foo;`, neither as a new type like `struct Foo(bar)`.
-
-Then impl `AggregateFunctionCreator` trait on it. The definition of the trait is:
-
-```Rust
-pub trait AggregateFunctionCreator: Send + Sync + Debug {
-    fn creator(&self) -> AccumulatorCreatorFunction;
-    fn output_type(&self) -> ConcreteDataType;
-    fn state_types(&self) -> Vec<ConcreteDataType>;
-}
-```
-
-You can use input data's type in methods that return output type and state types (just invoke `input_types()`).
-
-The output type is aggregate function's output data's type. For example, `SUM` aggregate function's output type is `u64` for a `u32` datatype column. The state types are accumulator's internal states' types. Take `AVG` aggregate function on a `i32` column as example, its state types are `i64` (for sum) and `u64` (for count).
-
-The `creator` function is where you define how an accumulator (that will be used in DataFusion) is created. You define "how" to create the accumulator (instead of "what" to create), using the input data's type as arguments. With input datatype known, you can create accumulator generically.
-
-# 2. Impl `Accumulator` trait for your accumulator.
-
-The accumulator is where you store the aggregate calculation states and evaluate a result. You must impl `Accumulator` trait for it. The trait's definition is:
-
-```Rust
-pub trait Accumulator: Send + Sync + Debug {
-    fn state(&self) -> Result<Vec<Value>>;
-    fn update_batch(&mut self, values: &[VectorRef]) -> Result<()>;
-    fn merge_batch(&mut self, states: &[VectorRef]) -> Result<()>;
-    fn evaluate(&self) -> Result<Value>;
-}
-```
-
-The DataFusion basically executes aggregate like this:
-
-1. Partitioning all input data for aggregate. Create an accumulator for each part.
-2. Call `update_batch` on each accumulator with partitioned data, to let you update your aggregate calculation.
-3. Call `state` to get each accumulator's internal state, the medial calculation result.
-4. Call `merge_batch` to merge all accumulator's internal state to one.
-5. Execute `evaluate` on the chosen one to get the final calculation result.
-
-Once you know the meaning of each method, you can easily write your accumulator. You can refer to `Median` accumulator or `SUM` accumulator defined in file `my_sum_udaf_example.rs` for more details.
-
-# 3. Register your aggregate function to our query engine.
-
-You can call `register_aggregate_function` method in query engine to register your aggregate function. To do that, you have to new an instance of struct `AggregateFunctionMeta`. The struct has three fields, first is the name of your aggregate function's name. The function name is case-sensitive due to DataFusion's restriction. We strongly recommend using lowercase for your name. If you have to use uppercase name, wrap your aggregate function with quotation marks. For example, if you define an aggregate function named "my_aggr", you can use "`SELECT MY_AGGR(x)`"; if you define "my_AGGR", you have to use "`SELECT "my_AGGR"(x)`".
-
-The second field is arg_counts ,the count of the arguments. Like accumulator `percentile`, calculating the p_number of the column. We need to input the value of column and the value of p to calculate, and so the count of the arguments is two.
-
-The third field is a function about how to create your accumulator creator that you defined in step 1 above. Create creator, that's a bit intertwined, but it is how we make DataFusion use a newly created aggregate function each time it executes a SQL, preventing the stored input types from affecting each other. The key detail can be starting looking at our `DfContextProviderAdapter` struct's `get_aggregate_meta` method.
-
-# (Optional) 4. Make your aggregate function automatically registered.
-
-If you've written a great aggregate function that wants to let everyone use it, you can make it automatically register to our query engine at start time. It's quick and simple, just refer to the `AggregateFunctions::register` function in `common/function/src/scalars/aggregate/mod.rs`.

docs/rfcs/2023-02-01-table-compaction.md

@@ -76,7 +76,7 @@ pub trait CompactionStrategy {
 ```
 
 The most suitable compaction strategy for time-series scenario would be 
-a hybrid strategy that combines time window compaction with size-tired compaction, just like [Cassandra](https://cassandra.apache.org/doc/latest/cassandra/operating/compaction/twcs.html) and [ScyllaDB](https://docs.scylladb.com/stable/architecture/compaction/compaction-strategies.html#time-window-compaction-strategy-twcs) does.
+a hybrid strategy that combines time window compaction with size-tired compaction, just like [Cassandra](https://cassandra.apache.org/doc/latest/cassandra/managing/operating/compaction/twcs.html) and [ScyllaDB](https://docs.scylladb.com/stable/architecture/compaction/compaction-strategies.html#time-window-compaction-strategy-twcs) does.
 
 We can first group SSTs in level n into buckets according to some predefined time window. Within that window, 
 SSTs are compacted in a size-tired manner (find SSTs with similar size and compact them to level n+1). 

docs/rfcs/2024-01-17-dataflow-framework.md

@@ -28,7 +28,7 @@ In order to do those things while maintaining a low memory footprint, you need t
 - Greptime Flow's is built on top of [Hydroflow](https://github.com/hydro-project/hydroflow).
 - We have three choices for the Dataflow/Streaming process framework for our simple continuous aggregation feature:
 1. Based on the timely/differential dataflow crate that [materialize](https://github.com/MaterializeInc/materialize) based on. Later, it's proved too obscure for a simple usage, and is hard to customize memory usage control.
-2. Based on a simple dataflow framework that we write from ground up, like what [arroyo](https://www.arroyo.dev/) or [risingwave](https://www.risingwave.dev/) did, for example the core streaming logic of [arroyo](https://github.com/ArroyoSystems/arroyo/blob/master/arroyo-datastream/src/lib.rs) only takes up to 2000 line of codes. However, it means maintaining another layer of dataflow framework, which might seem easy in the beginning, but I fear it might be too burdensome to maintain once we need more features.
+2. Based on a simple dataflow framework that we write from ground up, like what [arroyo](https://www.arroyo.dev/) or [risingwave](https://www.risingwave.dev/) did, for example the core streaming logic of [arroyo](https://github.com/ArroyoSystems/arroyo/blob/master/crates/arroyo-datastream/src/lib.rs) only takes up to 2000 line of codes. However, it means maintaining another layer of dataflow framework, which might seem easy in the beginning, but I fear it might be too burdensome to maintain once we need more features.
 3. Based on a simple and lower level dataflow framework that someone else write, like [hydroflow](https://github.com/hydro-project/hydroflow), this approach combines the best of both worlds. Firstly, it boasts ease of comprehension and customization. Secondly, the dataflow framework offers precisely the necessary features for crafting uncomplicated single-node dataflow programs while delivering decent performance.
 
 Hence, we choose the third option, and use a simple logical plan that's anagonistic to the underlying dataflow framework, as it only describe how the dataflow graph should be doing, not how it do that. And we built operator in hydroflow to execute the plan. And the result hydroflow graph is wrapped in a engine that only support data in/out and tick event to flush and compute the result. This provide a thin middle layer that's easy to maintain and allow switching to other dataflow framework if necessary.

docs/rfcs/2025-06-20-repartition.md

@@ -0,0 +1,154 @@
+---
+Feature Name: Repartition
+Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/6558
+Date: 2025-06-20
+Author: "Ruihang Xia <waynestxia@gmail.com>"
+---
+
+# Summary
+
+This RFC proposes a method for repartitioning a table, to adjust the partition rule and data distribution.
+
+# Motivation
+
+With time passing, the data distribution and skew pattern of a table might change. We need a way to repartition the table to suit the new pattern.
+
+# Details
+
+Here is a rough workflow diagram of the entire repartition process, each step is described in detail below.
+
+```mermaid
+sequenceDiagram
+    participant Frontend
+    participant Metasrv
+    participant Datanodes
+    participant Region0 as Region 0
+    
+    Frontend->>Frontend: Process request, validation etc.
+    Frontend->>Metasrv: Submit procedure
+    Metasrv->>Metasrv: Compute diff and generate migration plan
+    Metasrv->>Metasrv: Allocate necessary region resources (with Paas)
+    Metasrv->>Datanodes: Stop compaction and snapshot
+    rect rgb(255, 225, 225)
+        note over Frontend, Region0: No Ingestion Period
+        Metasrv->>Frontend: Stop processing write requests
+        Metasrv->>Metasrv: Update metadata
+        Metasrv->>Frontend: Start processing read requests
+    end
+    Metasrv->>Datanodes: Update region rule, stage version changes from now on
+    Region0->>Region0: Compute new manifests for all regions
+    Region0->>Datanodes: Submit manifest changes
+    Metasrv->>Datanodes: Recover compaction and snapshot, make staged changes visible
+
+    note over Frontend, Datanodes: Reload Cache
+    Metasrv->>Metasrv: Release resources (with Paas)
+    Metasrv->>Metasrv: Schedule optional compaction (to remote compactor)
+```
+
+## Preprocessing
+
+This phase is for static analysis of the new partition rule. The server can know whether the repartitioning is possible, how to do the repartitioning, and how much resources are needed.
+
+In theory, the input and output partition rules for repartitioning can be completely unrelated. But in practice, to avoid a very large change set, we'll only allow two simple kinds of change. One splits one region into two regions (region split) and another merges two regions into one (region merge).
+
+After validating the new partition rule using the same validation logic as table creation, we compute the difference between the old and new partition rules. The resulting diff may contain several independent groups of changes. During subsequent processing, each group of changes can be handled independently and can succeed or fail without affecting other groups or creating non-idempotently retryable scenarios.
+
+Next, we generate a repartition plan for each group of changes. Each plan contains this information for all regions involved in that particular plan. And one target region will only be referenced by a single plan.
+
+With those plans, we can determine the resource requirements for the repartition operation, where resources here primarily refer to Regions. Metasrv will coordinate with PaaS layer to pre-allocate the necessary regions at this stage. These new regions start completely empty, and their metadata and manifests will be populated during subsequent modification steps.
+
+## Data Processing
+
+This phase is primarily for region's change, including region's metadata (route table and the corresponding rule) and manifest.
+
+Once we start processing one plan through a procedure, we'll first stop the region's compaction and snapshot. This is to avoid any states being removed due to compaction (which may removes old SST files) and snapshot (which may removes old manifest files).
+
+Metasrv will trying to update the metadata of partition, or the region route table (related to `PartitionRuleManager`). This step is in the "no ingestion" scope, so no new data will be ingested. Since this won't take much time, the affection to the cluster is minimized. Metasrv will also update the region rule to corresponding regions on Datanodes. 
+
+Every regions and all the ingestion requests to the region server will have a version of region rule, to identify under which rule the request is processed. The version can be something like `hash(region_rule)`. Once the region rule on region server is updated, all ingestion request with old rule will be rejected, and all requests with new rule will be accepted but not visible. They can still be flushed to persisted storage, but their version change (new manifest) will be staged.
+
+Then region 0 (or let metasrv to pick any operational region) will compute the new manifests for all target regions. This step is done by first reading all old manifests, and remapping the files with new partition rule, to get the content of new manifests. Notice this step only handles the manifests before region rule change on region server, and won't touch those staged manifests, as they are already with the new rule.
+
+Those new manifest will be submitted to the corresponding target regions by region 0 via a `RegionEdit` request. If this request falls after a few retries, region 0 will try to rollback this change by directly overwriting the manifest on object storage. and report this failure to metasrv and let the entire repartition procedure to fail. And we can also optionally compute the new manifest for those staged version changes (like another repartition) and submit them to the target regions to make the also visible even if the repartition fails.
+
+In the other hand, a successful `RegionEdit` request also acknowledges those staged version changes and make them visible.
+
+After this step, the repartition is done in the data plane. We can start to process compaction and snapshot again.
+
+## Postprocessing
+
+After the main processing is done, we can do some extra postprocessing to reduce the performance impact of repartition. Including reloading caches in frontend's route table, metasrv's kv cache and datanode's read/write/page cache etc.
+
+We can also schedule an optional compaction to reorganize all the data file under the new partition rule to reduce potential fragmentation or read amplification.
+
+## Procedure
+
+Here describe the repartition procedure step by step:
+
+- <on frontend> Validating repartition request
+- <on frontend> Initialize the repartition procedure
+- Calculate rule diff and repartition plan group
+- Allocate necessary new regions
+- Lock the table key
+- For each repartition subprocedure
+  - Stop compaction and snapshot
+  - Forbid new ingestion requests, update metadata, allow ingestion requests.
+  - Update region rule to regions
+  - Pick one region to calculate new manifest for all regions in this repartition group
+  - Let that region to apply new manifest to each region via `RegionEdit`
+  - If failed after some retries, revert this manifest change to other succeeded regions and mark this failure.
+  - If all succeeded, acknowledge those staged version changes and make them visible.
+  - Return result
+- Collect results from subprocedure.
+  - For those who failed, we need to restart those regions to force reconstruct their status from manifests
+  - For those who succeeded, collect and merge their rule diff
+- Unlock the table key
+- Report the result to user.
+- <in background> Reload cache
+- <in background> Maybe trigger a special compaction
+
+In addition of sequential step, rollback is also an important part of this procedure. There are three steps can be rolled back when unrecoverable failure occurs.
+
+If the metadata update is not committed, we can overwrite the metadata to previous version. This step is scoped in the "no ingestion" period, so no new data will be ingested and the status of both datanode and metasrv will be consistent.
+
+If the `RegionEdit` to other regions is not acknowledged, or partial acknowledged, we can directly overwrite the manifest on object storage from the central region (who computes the new manifest), and force region server to reload corresponding region to load its state from object storage to recover.
+
+If the staged version changes are not acknowledged, we can re-compute manifest based on old rule for staged data, and apply them directly like above. This is like another smaller repartition for those staged data.
+
+## Region rule validation and diff calculation
+
+In the current codebase, the rule checker is not complete. It can't check uniqueness and completeness of the rule. This RFC also propose a new way to validate the rule.
+
+The proposed validation way is based on a check-point system, which first generates a group of check-points from the rule, and then check if all the point is covered and only covered by one rule.
+
+All the partition rule expressionis limited to be the form of `<column> <operator> <value>`, and the operator is limited to be comparison operators. Those expressions are allowed to be nested with `AND` and `OR` operators. Based on this, we can first extract all the unique values on each column, adding and subtracting a little epsilon to cover its left and right boundary.
+
+Since we accept integer, float and string as the value type, compute on them directly is not convenient. So we'll first normalize them to a common type and only need to preserve the relative partial ordering. This also avoids the problem of "what is next/previous value" of string and "what's a good precision" for float.
+
+After normalization, we get a set of scatter points for each column. Then we can generate a set of check-points by combining all the scatter points like building a cartesian product. This might bring a large number of check-points, so we can do an prune optimization to remove some of them by merging some of the expression zones. Those expressions who have identical N-1 edge sub-expressions with one adjacent edge can be merged together. This prune check is with a time complexity of O(N * M * log(M)), where N is the number of active dimensions and M is the number of expression zones. Diff calculation is also done by finding different expression zones between the old and new rule set, and check if we can transform one to another by merging some of the expression zones.
+
+The step to validate the check-points set against expressions can be treated as a tiny expression of `PhysicalExpr`. This evaluation will give a boolean matrix of K*M shape, where K is the number of check-points. We then check in each row of the matrix, if there is one and only one true value.
+
+## Compute and use new manifest
+
+We can generate a new set of manifest file based on old manifest and two versions of rule. From abvoe rule processing part, we can tell how a new rule & region is from previous one. So a simple way to get the new manifest is also apply the step of change to manifest files. E.g., if region A is from region B and C, we simply combine all file IDs from B and C to generate the content of A.
+
+If necessary, we can do this better by involving some metadata related to data, like min-max statistics of each file, and pre-evaluate over min-max to filter out unneeded files when generating new manifest.
+
+The way to use new manifest needs one more extra step based on the current implementation. We'll need to record either in manifest or in file metadata, of what rule is used when generating (flush or compaction) a SST file. Then in every single read request, we need to append the current region rule as predicate to the read request, to ensure no data belong to other regions will be read. We can use the stored region rule to reduce the number of new predicates to apply, by removing the identical predicate between the current region rule and the stored region rule. So ideally in a table that has not been repartitioned recently, the overhead of checking region rule is minimal.
+
+## Pre-required tasks
+
+In above steps, we assume some functionalities are implemented. Here list them with where they are used and how to implement them.
+
+### Cross-region read
+
+The current data directory structure is `{table_id}/{region_id}/[data/metadata]/{file_id}`, every region can only access files under their own directory. After repartition, data file may be placed in other previous old regions. So we need to support cross-region read. This new access method allows region to access any file under the same table. Related tracking issue is <https://github.com/GreptimeTeam/greptimedb/issues/6409>.
+
+### Global GC worker
+
+This is to simplify state management of data files. As one file may be referenced in multiple manifests, or no manifest at all. After this, every region and the repartition process only need to care about generateing and using new files, without tracking whether a file should be deleted or not. Leaving the deletion to the global GC worker. This worker basically works by counting reference from manifest file, and remove unused one. Related tracking issue is **TBD**.
+
+# Alternatives
+
+In the "Data Processing" section, we can enlarge the "no ingestion" period to include almost all the steps. This can simplify the entire procedure by a lot, but will bring a longer time of ingestion pause which may not be acceptable.

docs/rfcs/2025-07-04-compatibility-test-framework.md

@@ -0,0 +1,151 @@
+---
+Feature Name: Compatibility Test Framework
+Tracking Issue: TBD
+Date: 2025-07-04
+Author: "Ruihang Xia <waynestxia@gmail.com>"
+---
+
+# Summary
+
+This RFC proposes a compatibility test framework for GreptimeDB to ensure backward/forward compatibility for different versions of GreptimeDB.
+
+# Motivation
+
+In current practice, we don't have a systematic way to test and ensure the compatibility of different versions of GreptimeDB. Each time we release a new version, we need to manually test the compatibility with ad-hoc cases. This is not only time-consuming, but also prone to errors and unmaintainable. Highly rely on the release manager to ensure the compatibility of different versions of GreptimeDB.
+
+We don't have a detailed guide on the release SoP of how to test and ensure the compatibility of the new version. And has broken the compatibility of the new version many times (`v0.14.1` and `v0.15.1` are two examples, which are both released right after the major release).
+
+# Details
+
+This RFC proposes a compatibility test framework that is easy to maintain, extend and run. It can tell the compatibility between any given two versions of GreptimeDB, both backward and forward. It's based on the Sqlness library but used in a different way.
+
+Generally speaking, the framework is composed of two parts:
+
+1. Test cases: A set of test cases that are maintained dedicatedly for the compatibility test. Still in the `.sql` and `.result` format.
+2. Test framework: A new sqlness runner that is used to run the test cases. With some new features that is not required by the integration sqlness test.
+
+## Test Cases
+
+### Structure
+
+The case set is organized in three parts:
+
+- `1.feature`: Use a new feature
+- `2.verify`: Verify database behavior
+- `3.cleanup`: Paired with `1.feature`, cleanup the test environment.
+
+These three parts are organized in a tree structure, and should be run in sequence:
+
+```
+compatibility_test/
+├── 1.feature/
+│   ├── feature-a/
+│   ├── feature-b/
+│   └── feature-c/
+├── 2.verify/
+│   ├── verify-metadata/
+│   ├── verify-data/
+│   └── verify-schema/
+└── 3.cleanup/
+    ├── cleanup-a/
+    ├── cleanup-b/
+    └── cleanup-c/
+```
+
+### Example
+
+For example, for a new feature like adding new index option ([#6416](https://github.com/GreptimeTeam/greptimedb/pull/6416)), we (who implement the feature) create a new test case like this:
+
+```sql
+-- path: compatibility_test/1.feature/index-option/granularity_and_false_positive_rate.sql
+
+-- SQLNESS ARG since=0.15.0
+-- SQLNESS IGNORE_RESULT
+CREATE TABLE granularity_and_false_positive_rate (ts timestamp time index, val double) with ("index.granularity" = "8192", "index.false_positive_rate" = "0.01");
+```
+
+And
+
+```sql
+-- path: compatibility_test/3.cleanup/index-option/granularity_and_false_positive_rate.sql
+drop table granularity_and_false_positive_rate;
+```
+
+Since this new feature don't require some special way to verify the database behavior, we can reuse existing test cases in `2.verify/` to verify the database behavior. For example, we can reuse the `verify-metadata` test case to verify the metadata of the table.
+
+```sql
+-- path: compatibility_test/2.verify/verify-metadata/show-create-table.sql
+
+-- SQLNESS TEMPLATE TABLE="SHOW TABLES";
+SHOW CREATE TABLE $TABLE;
+```
+
+In this example, we use some new sqlness features that will be introduced in the next section (`since`, `IGNORE_RESULT`, `TEMPLATE`).
+
+### Maintenance
+
+Each time implement a new feature that should be covered by the compatibility test, we should create a new test case in `1.feature/` and `3.cleanup/` for them. And check if existing cases in `2.verify/` can be reused to verify the database behavior.
+
+This simulates an enthusiastic user who uses all the new features at the first time. All the new Maintenance burden is on the feature implementer to write one more test case for the new feature, to "fixation" the behavior. And once there is a breaking change in the future, it can be detected by the compatibility test framework automatically.
+
+Another topic is about deprecation. If a feature is deprecated, we should also mark it in the test case. Still use above example, assume we deprecate the `index.granularity` and `index.false_positive_rate` index options in `v0.99.0`, we can mark them as:
+```sql
+-- SQLNESS ARG since=0.15.0 till=0.99.0
+...
+```
+
+This tells the framework to ignore this feature in version `v0.99.0` and later. Currently, we have so many experimental features that are scheduled to be broken in the future, this is a good way to mark them.
+
+## Test Framework
+
+This section is about new sqlness features required by this framework.
+
+### Since and Till
+
+Follows the `ARG` interceptor in sqlness, we can mark a feature is available between two given versions. Only the `since` is required:
+
+```sql
+-- SQLNESS ARG since=VERSION_STRING [till=VERSION_STRING]
+```
+
+### IGNORE_RESULT
+
+`IGNORE_RESULT` is a new interceptor, it tells the runner to ignore the result of the query, only check whether the query is executed successfully.
+
+This is useful to reduce the Maintenance burden of the test cases, unlike the integration sqlness test, in most cases we don't care about the result of the query, only need to make sure the query is executed successfully.
+
+### TEMPLATE
+
+`TEMPLATE` is another new interceptor, it can generate queries from a template based on a runtime data.
+
+In above example, we need to run the `SHOW CREATE TABLE` query for all existing tables, so we can use the `TEMPLATE` interceptor to generate the query with a dynamic table list.
+
+### RUNNER
+
+There are also some extra requirement for the runner itself:
+
+- It should run the test cases in sequence, first `1.feature/`, then `2.verify/`, and finally `3.cleanup/`.
+- It should be able to fetch required version automatically to finish the test.
+- It should handle the `since` and `till` properly.
+
+On the `1.feature` phase, the runner needs to identify all features need to be tested by version number. And then restart with a new version (the `to` version) to run `2.verify/` and `3.cleanup/` phase.
+
+## Test Report
+
+Finally, we can run the compatibility test to verify the compatibility between any given two versions of GreptimeDB, for example:
+
+```bash
+# check backward compatibility between v0.15.0 and v0.16.0 when releasing v0.16.0
+./sqlness run --from=0.15.0 --to=0.16.0
+
+# check forward compatibility when downgrading from v0.15.0 to v0.13.0
+./sqlness run --from=0.15.0 --to=0.13.0
+```
+
+We can also use a script to run the compatibility test for all the versions in a given range to give a quick report with all versions we need.
+
+And we always bump the version in `Cargo.toml` to the next major release version, so the next major release version can be used as "latest" unpublished version for scenarios like local testing.
+
+# Alternatives
+
+There was a previous attempt to implement a compatibility test framework that was disabled due to some reasons [#3728](https://github.com/GreptimeTeam/greptimedb/issues/3728).

docs/rfcs/2025-07-23-global-gc-worker.md

@@ -0,0 +1,157 @@
+---
+Feature Name: "global-gc-worker"
+Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/6571
+Date: 2025-07-23
+Author: "discord9 <discord9@163.com>"
+---
+
+# Global GC Worker
+
+## Summary
+
+This RFC proposes the integration of a garbage collection (GC) mechanism within the Compaction process. This mechanism aims to manage and remove stale files that are no longer actively used by any system component, thereby reclaiming storage space.
+
+## Motivation
+
+With the introduction of features such as table repartitioning, a substantial number of Parquet files can become obsolete. Furthermore, failures during manifest updates may result in orphaned files that are never referenced by the system. Therefore, a periodic garbage collection mechanism is essential to reclaim storage space by systematically removing these unused files.
+
+## Details
+
+### Overview
+
+The garbage collection process will be integrated directly into the Compaction process. Upon the completion of a Compaction for a given region, the GC worker will be automatically triggered. Its primary function will be to identify and subsequently delete obsolete files that have persisted beyond their designated retention period. This integration ensures that garbage collection is performed in close conjunction with data lifecycle management, effectively leveraging the compaction process's inherent knowledge of file states.
+
+This design prioritizes correctness and safety by explicitly linking GC execution to a well-defined operational boundary: the successful completion of a compaction cycle.
+
+### Terminology
+
+- **Unused File**: Refers to a file present in the storage directory that has never been formally recorded in any manifest. A common scenario for this includes cases where a new SST file is successfully written to storage, but the subsequent update to the manifest fails, leaving the file unreferenced.
+- **Obsolete File**: Denotes a file that was previously recorded in a manifest but has since been explicitly marked for removal. This typically occurs following operations such as data repartitioning or compaction.
+
+### GC Worker Process
+
+The GC worker operates as an integral part of the Compaction process. Once a Compaction for a specific region is completed, the GC worker is automatically triggered. Executing this process on a `datanode` is preferred to eliminate the overhead associated with having to set object storage configurations in the `metasrv`.
+
+The detailed process is as follows:
+
+1.  **Invocation**: Upon the successful completion of a Compaction for a region, the GC worker is invoked.
+2.  **Manifest Reading**: The worker reads the region's primary manifest to obtain a comprehensive list of all files marked as obsolete. Concurrently, it reads any temporary manifests generated by long-running queries to identify files that are currently in active use, thereby preventing their premature deletion.
+3.  **Lingering Time Check (Obsolete Files)**: For each identified obsolete file, the GC worker evaluates its "lingering time." Which is the time passed after it had been removed from manifest.
+4.  **Deletion Marking (Obsolete Files)**: Files that have exceeded their maximum configurable lingering time and are not referenced by any active temporary manifests are marked for deletion.
+5.  **Lingering Time (Unused Files)**: Unused files (those never recorded in any manifest) are also subject to a configurable maximum lingering time before they are eligible for deletion.
+
+Following flowchart illustrates the GC worker's process:
+
+```mermaid
+flowchart TD
+    A[Compaction Completed] --> B[Trigger GC Worker]
+    B --> C[Scan Region Manifest]
+    C --> D[Identify File Types]
+    D --> E[Unused Files<br/>Never recorded in manifest]
+    D --> F[Obsolete Files<br/>Previously in manifest<br/>but marked for removal]
+    E --> G[Check Lingering Time]
+    F --> G
+    G --> H{File exceeds<br/>configured lingering time?}
+    H -->|No| I[Skip deletion]
+    H -->|Yes| J[Check Temporary Manifest]
+    J --> K{File in use by<br/>active queries?}
+    K -->|Yes| L[Retain file<br/>Wait for next GC cycle]
+    K -->|No| M[Safely delete file]
+    I --> N[End GC cycle]
+    L --> N
+    M --> O[Update Manifest]
+    O --> N
+    N --> P[Wait for next Compaction]
+    P --> A
+    style A fill:#e1f5fe
+    style B fill:#f3e5f5
+    style M fill:#e8f5e8
+    style L fill:#fff3e0
+```
+
+#### Handling Obsolete Files
+
+An obsolete file is permanently deleted only if two conditions are met:
+1. The time elapsed since its removal from the manifest (its obsolescence timestamp) exceeds a configurable threshold.
+2. It is not currently referenced by any active temporary manifests.
+
+
+#### Handling Unused Files
+
+With the integration of the GC worker into the Compaction process, the risk of accidentally deleting newly created SST files that have not yet been recorded in the manifest is significantly mitigated. Consequently, the concept of "Unused Files" as a distinct category primarily susceptible to accidental deletion is largely resolved. Any files that are genuinely "unused" (i.e., never referenced by any manifest, including temporary ones) can be safely deleted after a configurable maximum lingering time.
+
+For debugging and auditing purposes, a comprehensive list of recently deleted files can be maintained.
+
+### Ensuring Read Consistency
+
+To prevent the GC worker from inadvertently deleting files that are actively being utilized by long-running analytical queries, a robust protection mechanism is introduced. This mechanism relies on temporary manifests that are actively kept "alive" by the queries using them.
+
+When a long-running query is detected (e.g., by a slow query recorder), it will write a temporary manifest to the region's manifest directory. This manifest lists all files required for the query. However, simply creating this file is not enough, as a query runner might crash, leaving the temporary manifest orphaned and preventing garbage collection indefinitely.
+
+To address this, the following "heartbeat" mechanism is implemented:
+1.  **Periodic Updates**: The process executing the long-running query is responsible for periodically updating the modification timestamp of its temporary manifest file (i.e., "touching" the file). This serves as a heartbeat, signaling that the query is still active.
+2.  **GC Worker Verification**: When the GC worker runs, it scans for temporary manifests. For each one it finds, it checks the file's last modification time.
+3.  **Stale File Handling**: If a temporary manifest's last modification time is older than a configurable threshold, the GC worker considers it stale (left over from a crashed or terminated query). The GC worker will then delete this stale temporary manifest. Files that were protected only by this stale manifest are no longer shielded from garbage collection.
+
+This approach ensures that only files for genuinely active queries are protected. The lifecycle of the temporary manifest is managed dynamically: it is created when a long query starts, kept alive through periodic updates, and is either deleted by the query upon normal completion or automatically cleaned up by the GC worker if the query terminates unexpectedly.
+
+This mechanism may be too complex to implement at once. We can consider a two-phased approach:
+1.  **Phase 1 (Simple Time-Based Deletion)**: Initially, implement a simpler GC strategy that deletes obsolete files based solely on a configurable lingering time. This provides a baseline for space reclamation without the complexity of temporary manifests.
+2.  **Phase 2 (Consistency-Aware GC)**: Based on the practical effectiveness and observed issues from Phase 1, we can then decide whether to implement the full temporary manifest and heartbeat mechanism to handle long-running queries. This iterative approach allows for a quicker initial implementation while gathering real-world data to justify the need for a more complex solution.
+
+## Drawbacks
+
+- **Dependency on Compaction Frequency**: The integration of the GC worker with Compaction means that GC cycles are directly tied to the frequency of compactions. In environments with infrequent compaction operations, obsolete files may accumulate for extended periods before being reclaimed, potentially leading to increased storage consumption.
+- **Race Condition with Long-Running Queries**: A potential race condition exists if a long-running query initiates but haven't write its temporary manifest in time, while a compaction process simultaneously begins and marks files used by that query as obsolete. This scenario could lead to the premature deletion of files still required by the active query. To mitigate this, the threshold time for writing a temporary manifest should be significantly shorter than the lingering time configured for obsolete files, ensuring that next GC worker runs do not delete files that are now referenced by a temporary manifest if the query is still running.
+Also the read replica shouldn't be later in manifest version for more than the lingering time of obsolete files, otherwise it might ref to files that are already deleted by the GC worker.
+- need to upload tmp manifest to object storage, which may introduce additional complexity and potential performance overhead. But since long-running queries are typically not frequent, the performance impact is expected to be minimal.
+
+
+## Conclusion and Rationale
+
+This section summarizes the key aspects and trade-offs of the proposed integrated GC worker, highlighting its advantages and potential challenges.
+
+| Aspect | Current Proposal (Integrated GC) |
+| :--- | :--- |
+| **Implementation Complexity** | **Medium**. Requires careful integration with the compaction process and the slow query recorder for temporary manifest management. |
+| **Reliability** | **High**. Integration with compaction and leveraging temporary manifests from long-running queries significantly mitigates the risk of incorrect deletion. Accurate management of lingering times for obsolete files and prevention of accidental deletion of newly created SSTs enhance data safety. |
+| **Performance Overhead** | **Low to Medium**. The GC worker runs post-compaction, minimizing direct impact on write paths. Overhead from temporary manifest management by the slow query recorder is expected to be acceptable for long-running queries. |
+| **Impact on Other Components** | **Moderate**. Requires modifications to the compaction process to trigger GC and the slow query recorder to manage temporary manifests. This introduces some coupling but enhances overall data safety. |
+| **Deletion Strategy** | **State- and Time-Based**. Obsolete files are deleted based on a configurable lingering time, which is paused if the file is referenced by a temporary manifest. Unused files (never in a manifest) are also subject to a lingering time. |
+
+## Unresolved Questions and Future Work
+
+This section outlines key areas requiring further discussion and defines potential avenues for future development.
+
+*   **Slow Query Recorder Implementation**: Detailed specifications for modify slow query recorder's implementation and its precise interaction mechanisms with temporary manifests are needed.
+*   **Configurable Lingering Times**: Establish and make configurable the specific lingering times for both obsolete and unused files to optimize storage reclamation and data availability.
+
+## Alternatives
+
+### 1. Standalone GC Service
+
+Instead of integrating the GC worker directly into the Compaction process, a standalone GC service could be implemented. This service would operate independently, periodically scanning the storage for obsolete and unused files based on manifest information and predefined retention policies.
+
+**Pros:**
+*   **Decoupling**: Separates GC logic from compaction, allowing independent scaling and deployment.
+*   **Flexibility**: Can be configured to run at different frequencies and with different strategies than compaction.
+
+**Cons:**
+*   **Increased Complexity**: Requires a separate service to manage, monitor, and coordinate with other components.
+*   **Potential for Redundancy**: May duplicate some file scanning logic already present in compaction.
+*   **Consistency Challenges**: Ensuring read consistency would require more complex coordination mechanisms between the standalone GC service and active queries, potentially involving a distributed lock manager or a more sophisticated temporary manifest system.
+
+This alternative could be implemented in the future if the integrated GC worker proves insufficient or if there is a need for more advanced GC strategies.
+
+### 2. Manifest-Driven Deletion (No Lingering Time)
+
+This alternative would involve immediate deletion of files once they are removed from the manifest, without a lingering time.
+
+**Pros:**
+*   **Simplicity**: Simplifies the GC logic by removing the need for lingering time management.
+*   **Immediate Space Reclamation**: Storage space is reclaimed as soon as files are marked for deletion.
+
+**Cons:**
+*   **Increased Risk of Data Loss**: Higher risk of deleting files still in use by long-running queries or other processes if not perfectly synchronized.
+*   **Complex Read Consistency**: Requires extremely robust and immediate mechanisms to ensure that no active queries are referencing files marked for deletion, potentially leading to performance bottlenecks or complex error handling.
+*   **Debugging Challenges**: Difficult to debug issues related to premature file deletion due to the immediate nature of the operation.

flake.nix

@@ -15,8 +15,6 @@
       let
         pkgs = nixpkgs.legacyPackages.${system};
         buildInputs = with pkgs; [
-          libgit2
-          libz
         ];
         lib = nixpkgs.lib;
         rustToolchain = fenix.packages.${system}.fromToolchainName {

grafana/README.md

@@ -83,7 +83,7 @@ If you use the [Helm Chart](https://github.com/GreptimeTeam/helm-charts) to depl
 - `monitoring.enabled=true`: Deploys a standalone GreptimeDB instance dedicated to monitoring the cluster;
 - `grafana.enabled=true`: Deploys Grafana and automatically imports the monitoring dashboard;
 
-The standalone GreptimeDB instance will collect metrics from your cluster, and the dashboard will be available in the Grafana UI. For detailed deployment instructions, please refer to our [Kubernetes deployment guide](https://docs.greptime.com/user-guide/deployments-administration-administration/deploy-on-kubernetes/getting-started).
+The standalone GreptimeDB instance will collect metrics from your cluster, and the dashboard will be available in the Grafana UI. For detailed deployment instructions, please refer to our [Kubernetes deployment guide](https://docs.greptime.com/user-guide/deployments-administration/deploy-on-kubernetes/overview).
 
 ### Self-host Prometheus and import dashboards manually
 

grafana/dashboards/metrics/cluster/dashboard.md

@@ -21,14 +21,14 @@
 # Resources
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Datanode Memory per Instance | `sum(process_resident_memory_bytes{instance=~"$datanode"}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `prometheus` | `decbytes` | `[{{instance}}]-[{{ pod }}]` |
-| Datanode CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{instance=~"$datanode"}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
-| Frontend Memory per Instance | `sum(process_resident_memory_bytes{instance=~"$frontend"}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `prometheus` | `decbytes` | `[{{ instance }}]-[{{ pod }}]` |
-| Frontend CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{instance=~"$frontend"}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]-cpu` |
-| Metasrv Memory per Instance | `sum(process_resident_memory_bytes{instance=~"$metasrv"}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `prometheus` | `decbytes` | `[{{ instance }}]-[{{ pod }}]-resident` |
-| Metasrv CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{instance=~"$metasrv"}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
-| Flownode Memory per Instance | `sum(process_resident_memory_bytes{instance=~"$flownode"}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `prometheus` | `decbytes` | `[{{ instance }}]-[{{ pod }}]` |
-| Flownode CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{instance=~"$flownode"}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
+| Datanode Memory per Instance | `sum(process_resident_memory_bytes{instance=~"$datanode"}) by (instance, pod)`<br/>`max(greptime_memory_limit_in_bytes{app="greptime-datanode"})` | `timeseries` | Current memory usage by instance | `prometheus` | `bytes` | `[{{instance}}]-[{{ pod }}]` |
+| Datanode CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{instance=~"$datanode"}[$__rate_interval]) * 1000) by (instance, pod)`<br/>`max(greptime_cpu_limit_in_millicores{app="greptime-datanode"})` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
+| Frontend Memory per Instance | `sum(process_resident_memory_bytes{instance=~"$frontend"}) by (instance, pod)`<br/>`max(greptime_memory_limit_in_bytes{app="greptime-frontend"})` | `timeseries` | Current memory usage by instance | `prometheus` | `bytes` | `[{{ instance }}]-[{{ pod }}]` |
+| Frontend CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{instance=~"$frontend"}[$__rate_interval]) * 1000) by (instance, pod)`<br/>`max(greptime_cpu_limit_in_millicores{app="greptime-frontend"})` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]-cpu` |
+| Metasrv Memory per Instance | `sum(process_resident_memory_bytes{instance=~"$metasrv"}) by (instance, pod)`<br/>`max(greptime_memory_limit_in_bytes{app="greptime-metasrv"})` | `timeseries` | Current memory usage by instance | `prometheus` | `bytes` | `[{{ instance }}]-[{{ pod }}]-resident` |
+| Metasrv CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{instance=~"$metasrv"}[$__rate_interval]) * 1000) by (instance, pod)`<br/>`max(greptime_cpu_limit_in_millicores{app="greptime-metasrv"})` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
+| Flownode Memory per Instance | `sum(process_resident_memory_bytes{instance=~"$flownode"}) by (instance, pod)`<br/>`max(greptime_memory_limit_in_bytes{app="greptime-flownode"})` | `timeseries` | Current memory usage by instance | `prometheus` | `bytes` | `[{{ instance }}]-[{{ pod }}]` |
+| Flownode CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{instance=~"$flownode"}[$__rate_interval]) * 1000) by (instance, pod)`<br/>`max(greptime_cpu_limit_in_millicores{app="greptime-flownode"})` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
 # Frontend Requests
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
@@ -72,18 +72,19 @@
 | 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(instance, stage, pod) (rate(greptime_region_worker_handle_write_sum[$__rate_interval]))/sum by(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` |
 | Active Series and Field Builders Count | `sum by(instance, pod) (greptime_mito_memtable_active_series_count)`<br/>`sum by(instance, pod) (greptime_mito_memtable_field_builder_count)` | `timeseries` | Compaction oinput output bytes | `prometheus` | `none` | `[{{instance}}]-[{{pod}}]-series` |
 | 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` |
+| Cache Miss | `sum by (instance,pod, type) (rate(greptime_mito_cache_miss{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | The local cache miss of the datanode. | `prometheus` | -- | `[{{instance}}]-[{{pod}}]-[{{type}}]` |
 # OpenDAL
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
 | QPS per Instance | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
-| Read QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation="read"}[$__rate_interval]))` | `timeseries` | Read QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
-| Read P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode",operation="read"}[$__rate_interval])))` | `timeseries` | Read P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-{{scheme}}` |
-| Write QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation="write"}[$__rate_interval]))` | `timeseries` | Write QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-{{scheme}}` |
-| Write P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation="write"}[$__rate_interval])))` | `timeseries` | Write P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
+| Read QPS per Instance | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation=~"read\|Reader::read"}[$__rate_interval]))` | `timeseries` | Read QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
+| Read P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode",operation=~"read\|Reader::read"}[$__rate_interval])))` | `timeseries` | Read P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
+| Write QPS per Instance | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation=~"write\|Writer::write\|Writer::close"}[$__rate_interval]))` | `timeseries` | Write QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
+| Write P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation =~ "Writer::write\|Writer::close\|write"}[$__rate_interval])))` | `timeseries` | Write P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
 | List QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation="list"}[$__rate_interval]))` | `timeseries` | List QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
 | List P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation="list"}[$__rate_interval])))` | `timeseries` | List P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
 | Other Requests per Instance | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode",operation!~"read\|write\|list\|stat"}[$__rate_interval]))` | `timeseries` | Other Requests per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
-| Other Request P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation!~"read\|write\|list"}[$__rate_interval])))` | `timeseries` | Other Request P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
+| Other Request P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation!~"read\|write\|list\|Writer::write\|Writer::close\|Reader::read"}[$__rate_interval])))` | `timeseries` | Other Request P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
 | Opendal traffic | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_bytes_sum{instance=~"$datanode"}[$__rate_interval]))` | `timeseries` | Total traffic as in bytes by instance and operation | `prometheus` | `decbytes` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
 | OpenDAL errors per Instance | `sum by(instance, pod, scheme, operation, error) (rate(opendal_operation_errors_total{instance=~"$datanode", error!="NotFound"}[$__rate_interval]))` | `timeseries` | OpenDAL error counts per Instance. | `prometheus` | -- | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]-[{{error}}]` |
 # Metasrv

grafana/dashboards/metrics/cluster/dashboard.yaml

@@ -180,13 +180,18 @@ groups:
         - title: Datanode Memory per Instance
           type: timeseries
           description: Current memory usage by instance
-          unit: decbytes
+          unit: bytes
           queries:
             - expr: sum(process_resident_memory_bytes{instance=~"$datanode"}) by (instance, pod)
               datasource:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{instance}}]-[{{ pod }}]'
+            - expr: max(greptime_memory_limit_in_bytes{app="greptime-datanode"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: limit
         - title: Datanode CPU Usage per Instance
           type: timeseries
           description: Current cpu usage by instance
@@ -197,16 +202,26 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{ instance }}]-[{{ pod }}]'
+            - expr: max(greptime_cpu_limit_in_millicores{app="greptime-datanode"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: limit
         - title: Frontend Memory per Instance
           type: timeseries
           description: Current memory usage by instance
-          unit: decbytes
+          unit: bytes
           queries:
             - expr: sum(process_resident_memory_bytes{instance=~"$frontend"}) by (instance, pod)
               datasource:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{ instance }}]-[{{ pod }}]'
+            - expr: max(greptime_memory_limit_in_bytes{app="greptime-frontend"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: limit
         - title: Frontend CPU Usage per Instance
           type: timeseries
           description: Current cpu usage by instance
@@ -217,16 +232,26 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{ instance }}]-[{{ pod }}]-cpu'
+            - expr: max(greptime_cpu_limit_in_millicores{app="greptime-frontend"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: limit
         - title: Metasrv Memory per Instance
           type: timeseries
           description: Current memory usage by instance
-          unit: decbytes
+          unit: bytes
           queries:
             - expr: sum(process_resident_memory_bytes{instance=~"$metasrv"}) by (instance, pod)
               datasource:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{ instance }}]-[{{ pod }}]-resident'
+            - expr: max(greptime_memory_limit_in_bytes{app="greptime-metasrv"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: limit
         - title: Metasrv CPU Usage per Instance
           type: timeseries
           description: Current cpu usage by instance
@@ -237,16 +262,26 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{ instance }}]-[{{ pod }}]'
+            - expr: max(greptime_cpu_limit_in_millicores{app="greptime-metasrv"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: limit
         - title: Flownode Memory per Instance
           type: timeseries
           description: Current memory usage by instance
-          unit: decbytes
+          unit: bytes
           queries:
             - expr: sum(process_resident_memory_bytes{instance=~"$flownode"}) by (instance, pod)
               datasource:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{ instance }}]-[{{ pod }}]'
+            - expr: max(greptime_memory_limit_in_bytes{app="greptime-flownode"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: limit
         - title: Flownode CPU Usage per Instance
           type: timeseries
           description: Current cpu usage by instance
@@ -257,6 +292,11 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{ instance }}]-[{{ pod }}]'
+            - expr: max(greptime_cpu_limit_in_millicores{app="greptime-flownode"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: limit
     - title: Frontend Requests
       panels:
         - title: HTTP QPS per Instance
@@ -642,6 +682,15 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-AVG'
+        - title: Cache Miss
+          type: timeseries
+          description: The local cache miss of the datanode.
+          queries:
+            - expr: sum by (instance,pod, type) (rate(greptime_mito_cache_miss{instance=~"$datanode"}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{type}}]'
     - title: OpenDAL
       panels:
         - title: QPS per Instance
@@ -659,41 +708,41 @@ groups:
           description: Read QPS per Instance.
           unit: ops
           queries:
-            - expr: sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation="read"}[$__rate_interval]))
+            - expr: sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation=~"read|Reader::read"}[$__rate_interval]))
               datasource:
                 type: prometheus
                 uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]'
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]'
         - title: Read P99 per Instance
           type: timeseries
           description: Read P99 per Instance.
           unit: s
           queries:
-            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode",operation="read"}[$__rate_interval])))
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode",operation=~"read|Reader::read"}[$__rate_interval])))
               datasource:
                 type: prometheus
                 uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-{{scheme}}'
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]'
         - title: Write QPS per Instance
           type: timeseries
           description: Write QPS per Instance.
           unit: ops
           queries:
-            - expr: sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation="write"}[$__rate_interval]))
+            - expr: sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{instance=~"$datanode", operation=~"write|Writer::write|Writer::close"}[$__rate_interval]))
               datasource:
                 type: prometheus
                 uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-{{scheme}}'
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]'
         - title: Write P99 per Instance
           type: timeseries
           description: Write P99 per Instance.
           unit: s
           queries:
-            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation="write"}[$__rate_interval])))
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation =~ "Writer::write|Writer::close|write"}[$__rate_interval])))
               datasource:
                 type: prometheus
                 uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]'
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]'
         - title: List QPS per Instance
           type: timeseries
           description: List QPS per Instance.
@@ -729,7 +778,7 @@ groups:
           description: Other Request P99 per Instance.
           unit: s
           queries:
-            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation!~"read|write|list"}[$__rate_interval])))
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{instance=~"$datanode", operation!~"read|write|list|Writer::write|Writer::close|Reader::read"}[$__rate_interval])))
               datasource:
                 type: prometheus
                 uid: ${metrics}

grafana/dashboards/metrics/standalone/dashboard.md

@@ -21,14 +21,14 @@
 # Resources
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
-| Datanode Memory per Instance | `sum(process_resident_memory_bytes{}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `prometheus` | `decbytes` | `[{{instance}}]-[{{ pod }}]` |
-| Datanode CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
-| Frontend Memory per Instance | `sum(process_resident_memory_bytes{}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `prometheus` | `decbytes` | `[{{ instance }}]-[{{ pod }}]` |
-| Frontend CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]-cpu` |
-| Metasrv Memory per Instance | `sum(process_resident_memory_bytes{}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `prometheus` | `decbytes` | `[{{ instance }}]-[{{ pod }}]-resident` |
-| Metasrv CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
-| Flownode Memory per Instance | `sum(process_resident_memory_bytes{}) by (instance, pod)` | `timeseries` | Current memory usage by instance | `prometheus` | `decbytes` | `[{{ instance }}]-[{{ pod }}]` |
-| Flownode CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
+| Datanode Memory per Instance | `sum(process_resident_memory_bytes{}) by (instance, pod)`<br/>`max(greptime_memory_limit_in_bytes{app="greptime-datanode"})` | `timeseries` | Current memory usage by instance | `prometheus` | `bytes` | `[{{instance}}]-[{{ pod }}]` |
+| Datanode CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)`<br/>`max(greptime_cpu_limit_in_millicores{app="greptime-datanode"})` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
+| Frontend Memory per Instance | `sum(process_resident_memory_bytes{}) by (instance, pod)`<br/>`max(greptime_memory_limit_in_bytes{app="greptime-frontend"})` | `timeseries` | Current memory usage by instance | `prometheus` | `bytes` | `[{{ instance }}]-[{{ pod }}]` |
+| Frontend CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)`<br/>`max(greptime_cpu_limit_in_millicores{app="greptime-frontend"})` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]-cpu` |
+| Metasrv Memory per Instance | `sum(process_resident_memory_bytes{}) by (instance, pod)`<br/>`max(greptime_memory_limit_in_bytes{app="greptime-metasrv"})` | `timeseries` | Current memory usage by instance | `prometheus` | `bytes` | `[{{ instance }}]-[{{ pod }}]-resident` |
+| Metasrv CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)`<br/>`max(greptime_cpu_limit_in_millicores{app="greptime-metasrv"})` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
+| Flownode Memory per Instance | `sum(process_resident_memory_bytes{}) by (instance, pod)`<br/>`max(greptime_memory_limit_in_bytes{app="greptime-flownode"})` | `timeseries` | Current memory usage by instance | `prometheus` | `bytes` | `[{{ instance }}]-[{{ pod }}]` |
+| Flownode CPU Usage per Instance | `sum(rate(process_cpu_seconds_total{}[$__rate_interval]) * 1000) by (instance, pod)`<br/>`max(greptime_cpu_limit_in_millicores{app="greptime-flownode"})` | `timeseries` | Current cpu usage by instance | `prometheus` | `none` | `[{{ instance }}]-[{{ pod }}]` |
 # Frontend Requests
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
@@ -72,18 +72,19 @@
 | 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(instance, stage, pod) (rate(greptime_region_worker_handle_write_sum[$__rate_interval]))/sum by(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` |
 | Active Series and Field Builders Count | `sum by(instance, pod) (greptime_mito_memtable_active_series_count)`<br/>`sum by(instance, pod) (greptime_mito_memtable_field_builder_count)` | `timeseries` | Compaction oinput output bytes | `prometheus` | `none` | `[{{instance}}]-[{{pod}}]-series` |
 | 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` |
+| Cache Miss | `sum by (instance,pod, type) (rate(greptime_mito_cache_miss{}[$__rate_interval]))` | `timeseries` | The local cache miss of the datanode. | `prometheus` | -- | `[{{instance}}]-[{{pod}}]-[{{type}}]` |
 # OpenDAL
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |
 | --- | --- | --- | --- | --- | --- | --- |
 | QPS per Instance | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{}[$__rate_interval]))` | `timeseries` | QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
-| Read QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{ operation="read"}[$__rate_interval]))` | `timeseries` | Read QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
-| Read P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{operation="read"}[$__rate_interval])))` | `timeseries` | Read P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-{{scheme}}` |
-| Write QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{ operation="write"}[$__rate_interval]))` | `timeseries` | Write QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-{{scheme}}` |
-| Write P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{ operation="write"}[$__rate_interval])))` | `timeseries` | Write P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
+| Read QPS per Instance | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{ operation=~"read\|Reader::read"}[$__rate_interval]))` | `timeseries` | Read QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
+| Read P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{operation=~"read\|Reader::read"}[$__rate_interval])))` | `timeseries` | Read P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
+| Write QPS per Instance | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{ operation=~"write\|Writer::write\|Writer::close"}[$__rate_interval]))` | `timeseries` | Write QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
+| Write P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{ operation =~ "Writer::write\|Writer::close\|write"}[$__rate_interval])))` | `timeseries` | Write P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
 | List QPS per Instance | `sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{ operation="list"}[$__rate_interval]))` | `timeseries` | List QPS per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
 | List P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{ operation="list"}[$__rate_interval])))` | `timeseries` | List P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]` |
 | Other Requests per Instance | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{operation!~"read\|write\|list\|stat"}[$__rate_interval]))` | `timeseries` | Other Requests per Instance. | `prometheus` | `ops` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
-| Other Request P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{ operation!~"read\|write\|list"}[$__rate_interval])))` | `timeseries` | Other Request P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
+| Other Request P99 per Instance | `histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{ operation!~"read\|write\|list\|Writer::write\|Writer::close\|Reader::read"}[$__rate_interval])))` | `timeseries` | Other Request P99 per Instance. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
 | Opendal traffic | `sum by(instance, pod, scheme, operation) (rate(opendal_operation_bytes_sum{}[$__rate_interval]))` | `timeseries` | Total traffic as in bytes by instance and operation | `prometheus` | `decbytes` | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]` |
 | OpenDAL errors per Instance | `sum by(instance, pod, scheme, operation, error) (rate(opendal_operation_errors_total{ error!="NotFound"}[$__rate_interval]))` | `timeseries` | OpenDAL error counts per Instance. | `prometheus` | -- | `[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]-[{{error}}]` |
 # Metasrv

grafana/dashboards/metrics/standalone/dashboard.yaml

@@ -180,13 +180,18 @@ groups:
         - title: Datanode Memory per Instance
           type: timeseries
           description: Current memory usage by instance
-          unit: decbytes
+          unit: bytes
           queries:
             - expr: sum(process_resident_memory_bytes{}) by (instance, pod)
               datasource:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{instance}}]-[{{ pod }}]'
+            - expr: max(greptime_memory_limit_in_bytes{app="greptime-datanode"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: limit
         - title: Datanode CPU Usage per Instance
           type: timeseries
           description: Current cpu usage by instance
@@ -197,16 +202,26 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{ instance }}]-[{{ pod }}]'
+            - expr: max(greptime_cpu_limit_in_millicores{app="greptime-datanode"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: limit
         - title: Frontend Memory per Instance
           type: timeseries
           description: Current memory usage by instance
-          unit: decbytes
+          unit: bytes
           queries:
             - expr: sum(process_resident_memory_bytes{}) by (instance, pod)
               datasource:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{ instance }}]-[{{ pod }}]'
+            - expr: max(greptime_memory_limit_in_bytes{app="greptime-frontend"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: limit
         - title: Frontend CPU Usage per Instance
           type: timeseries
           description: Current cpu usage by instance
@@ -217,16 +232,26 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{ instance }}]-[{{ pod }}]-cpu'
+            - expr: max(greptime_cpu_limit_in_millicores{app="greptime-frontend"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: limit
         - title: Metasrv Memory per Instance
           type: timeseries
           description: Current memory usage by instance
-          unit: decbytes
+          unit: bytes
           queries:
             - expr: sum(process_resident_memory_bytes{}) by (instance, pod)
               datasource:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{ instance }}]-[{{ pod }}]-resident'
+            - expr: max(greptime_memory_limit_in_bytes{app="greptime-metasrv"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: limit
         - title: Metasrv CPU Usage per Instance
           type: timeseries
           description: Current cpu usage by instance
@@ -237,16 +262,26 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{ instance }}]-[{{ pod }}]'
+            - expr: max(greptime_cpu_limit_in_millicores{app="greptime-metasrv"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: limit
         - title: Flownode Memory per Instance
           type: timeseries
           description: Current memory usage by instance
-          unit: decbytes
+          unit: bytes
           queries:
             - expr: sum(process_resident_memory_bytes{}) by (instance, pod)
               datasource:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{ instance }}]-[{{ pod }}]'
+            - expr: max(greptime_memory_limit_in_bytes{app="greptime-flownode"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: limit
         - title: Flownode CPU Usage per Instance
           type: timeseries
           description: Current cpu usage by instance
@@ -257,6 +292,11 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{ instance }}]-[{{ pod }}]'
+            - expr: max(greptime_cpu_limit_in_millicores{app="greptime-flownode"})
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: limit
     - title: Frontend Requests
       panels:
         - title: HTTP QPS per Instance
@@ -642,6 +682,15 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-AVG'
+        - title: Cache Miss
+          type: timeseries
+          description: The local cache miss of the datanode.
+          queries:
+            - expr: sum by (instance,pod, type) (rate(greptime_mito_cache_miss{}[$__rate_interval]))
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{type}}]'
     - title: OpenDAL
       panels:
         - title: QPS per Instance
@@ -659,41 +708,41 @@ groups:
           description: Read QPS per Instance.
           unit: ops
           queries:
-            - expr: sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{ operation="read"}[$__rate_interval]))
+            - expr: sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{ operation=~"read|Reader::read"}[$__rate_interval]))
               datasource:
                 type: prometheus
                 uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]'
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]'
         - title: Read P99 per Instance
           type: timeseries
           description: Read P99 per Instance.
           unit: s
           queries:
-            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{operation="read"}[$__rate_interval])))
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{operation=~"read|Reader::read"}[$__rate_interval])))
               datasource:
                 type: prometheus
                 uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-{{scheme}}'
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]'
         - title: Write QPS per Instance
           type: timeseries
           description: Write QPS per Instance.
           unit: ops
           queries:
-            - expr: sum by(instance, pod, scheme) (rate(opendal_operation_duration_seconds_count{ operation="write"}[$__rate_interval]))
+            - expr: sum by(instance, pod, scheme, operation) (rate(opendal_operation_duration_seconds_count{ operation=~"write|Writer::write|Writer::close"}[$__rate_interval]))
               datasource:
                 type: prometheus
                 uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-{{scheme}}'
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]'
         - title: Write P99 per Instance
           type: timeseries
           description: Write P99 per Instance.
           unit: s
           queries:
-            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme) (rate(opendal_operation_duration_seconds_bucket{ operation="write"}[$__rate_interval])))
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{ operation =~ "Writer::write|Writer::close|write"}[$__rate_interval])))
               datasource:
                 type: prometheus
                 uid: ${metrics}
-              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]'
+              legendFormat: '[{{instance}}]-[{{pod}}]-[{{scheme}}]-[{{operation}}]'
         - title: List QPS per Instance
           type: timeseries
           description: List QPS per Instance.
@@ -729,7 +778,7 @@ groups:
           description: Other Request P99 per Instance.
           unit: s
           queries:
-            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{ operation!~"read|write|list"}[$__rate_interval])))
+            - expr: histogram_quantile(0.99, sum by(instance, pod, le, scheme, operation) (rate(opendal_operation_duration_seconds_bucket{ operation!~"read|write|list|Writer::write|Writer::close|Reader::read"}[$__rate_interval])))
               datasource:
                 type: prometheus
                 uid: ${metrics}

grafana/scripts/check.sh

@@ -26,7 +26,7 @@ check_dashboards_generation() {
   ./grafana/scripts/gen-dashboards.sh
 
   if [[ -n "$(git diff --name-only grafana/dashboards/metrics)" ]]; then
-    echo "Error: The dashboards are not generated correctly. You should execute the `make dashboards` command."
+    echo "Error: The dashboards are not generated correctly. You should execute the 'make dashboards' command."
     exit 1
   fi
 }

licenserc.toml

@@ -29,9 +29,11 @@ excludes = [
     # enterprise
     "src/common/meta/src/rpc/ddl/trigger.rs",
     "src/operator/src/expr_helper/trigger.rs",
+    "src/sql/src/statements/alter/trigger.rs",
     "src/sql/src/statements/create/trigger.rs",
     "src/sql/src/statements/show/trigger.rs",
     "src/sql/src/statements/drop/trigger.rs",
+    "src/sql/src/parsers/alter_parser/trigger.rs",
     "src/sql/src/parsers/create_parser/trigger.rs",
     "src/sql/src/parsers/show_parser/trigger.rs",
     "src/mito2/src/extension.rs",

scripts/check-super-imports.py

@@ -13,8 +13,8 @@
 # limitations under the License.
 
 import os
-import re
 from multiprocessing import Pool
+from pathlib import Path
 
 
 def find_rust_files(directory):
@@ -24,6 +24,10 @@ def find_rust_files(directory):
         if "test" in root.lower():
             continue
 
+        # Skip the target directory
+        if "target" in Path(root).parts:
+            continue
+
         for file in files:
             # Skip files with "test" in the filename
             if "test" in file.lower():

scripts/install.sh

@@ -53,6 +53,54 @@ get_arch_type() {
   esac
 }
 
+# Verify SHA256 checksum
+verify_sha256() {
+  file="$1"
+  expected_sha256="$2"
+
+  if command -v sha256sum >/dev/null 2>&1; then
+    actual_sha256=$(sha256sum "$file" | cut -d' ' -f1)
+  elif command -v shasum >/dev/null 2>&1; then
+    actual_sha256=$(shasum -a 256 "$file" | cut -d' ' -f1)
+  else
+    echo "Warning: No SHA256 verification tool found (sha256sum or shasum). Skipping checksum verification."
+    return 0
+  fi
+
+  if [ "$actual_sha256" = "$expected_sha256" ]; then
+    echo "SHA256 checksum verified successfully."
+    return 0
+  else
+    echo "Error: SHA256 checksum verification failed!"
+    echo "Expected: $expected_sha256"
+    echo "Actual: $actual_sha256"
+    return 1
+  fi
+}
+
+# Prompt for user confirmation (compatible with different shells)
+prompt_confirmation() {
+  message="$1"
+  printf "%s (y/N): " "$message"
+
+  # Try to read user input, fallback if read fails
+  answer=""
+  if read answer </dev/tty 2>/dev/null; then
+    case "$answer" in
+      [Yy]|[Yy][Ee][Ss])
+        return 0
+        ;;
+      *)
+        return 1
+        ;;
+    esac
+  else
+    echo ""
+    echo "Cannot read user input. Defaulting to No."
+    return 1
+  fi
+}
+
 download_artifact() {
   if [ -n "${OS_TYPE}" ] && [ -n "${ARCH_TYPE}" ]; then
     # Use the latest stable released version.
@@ -71,17 +119,104 @@ download_artifact() {
     fi
 
     echo "Downloading ${BIN}, OS: ${OS_TYPE}, Arch: ${ARCH_TYPE}, Version: ${VERSION}"
-    PACKAGE_NAME="${BIN}-${OS_TYPE}-${ARCH_TYPE}-${VERSION}.tar.gz"
+    PKG_NAME="${BIN}-${OS_TYPE}-${ARCH_TYPE}-${VERSION}"
+    PACKAGE_NAME="${PKG_NAME}.tar.gz"
+    SHA256_FILE="${PKG_NAME}.sha256sum"
 
     if [ -n "${PACKAGE_NAME}" ]; then
-      wget "https://github.com/${GITHUB_ORG}/${GITHUB_REPO}/releases/download/${VERSION}/${PACKAGE_NAME}"
+      # Check if files already exist and prompt for override
+      if [ -f "${PACKAGE_NAME}" ]; then
+        echo "File ${PACKAGE_NAME} already exists."
+        if prompt_confirmation "Do you want to override it?"; then
+          echo "Overriding existing file..."
+          rm -f "${PACKAGE_NAME}"
+        else
+          echo "Skipping download. Using existing file."
+        fi
+      fi
+
+      if [ -f "${BIN}" ]; then
+        echo "Binary ${BIN} already exists."
+        if prompt_confirmation "Do you want to override it?"; then
+          echo "Will override existing binary..."
+          rm -f "${BIN}"
+        else
+          echo "Installation cancelled."
+          exit 0
+        fi
+      fi
+
+      # Download package if not exists
+      if [ ! -f "${PACKAGE_NAME}" ]; then
+        echo "Downloading ${PACKAGE_NAME}..."
+        # Use curl instead of wget for better compatibility
+        if command -v curl >/dev/null 2>&1; then
+          if ! curl -L -o "${PACKAGE_NAME}" "https://github.com/${GITHUB_ORG}/${GITHUB_REPO}/releases/download/${VERSION}/${PACKAGE_NAME}"; then
+            echo "Error: Failed to download ${PACKAGE_NAME}"
+            exit 1
+          fi
+        elif command -v wget >/dev/null 2>&1; then
+          if ! wget -O "${PACKAGE_NAME}" "https://github.com/${GITHUB_ORG}/${GITHUB_REPO}/releases/download/${VERSION}/${PACKAGE_NAME}"; then
+            echo "Error: Failed to download ${PACKAGE_NAME}"
+            exit 1
+          fi
+        else
+          echo "Error: Neither curl nor wget is available for downloading."
+          exit 1
+        fi
+      fi
+
+      # Download and verify SHA256 checksum
+      echo "Downloading SHA256 checksum..."
+      sha256_download_success=0
+      if command -v curl >/dev/null 2>&1; then
+        if curl -L -s -o "${SHA256_FILE}" "https://github.com/${GITHUB_ORG}/${GITHUB_REPO}/releases/download/${VERSION}/${SHA256_FILE}" 2>/dev/null; then
+          sha256_download_success=1
+        fi
+      elif command -v wget >/dev/null 2>&1; then
+        if wget -q -O "${SHA256_FILE}" "https://github.com/${GITHUB_ORG}/${GITHUB_REPO}/releases/download/${VERSION}/${SHA256_FILE}" 2>/dev/null; then
+          sha256_download_success=1
+        fi
+      fi
+
+      if [ $sha256_download_success -eq 1 ] && [ -f "${SHA256_FILE}" ]; then
+        expected_sha256=$(cat "${SHA256_FILE}" | cut -d' ' -f1)
+        if [ -n "$expected_sha256" ]; then
+          if ! verify_sha256 "${PACKAGE_NAME}" "${expected_sha256}"; then
+            echo "SHA256 verification failed. Removing downloaded file."
+            rm -f "${PACKAGE_NAME}" "${SHA256_FILE}"
+            exit 1
+          fi
+        else
+          echo "Warning: Could not parse SHA256 checksum from file."
+        fi
+        rm -f "${SHA256_FILE}"
+      else
+        echo "Warning: Could not download SHA256 checksum file. Skipping verification."
+      fi
 
       # Extract the binary and clean the rest.
-      tar xvf "${PACKAGE_NAME}" && \
-      mv "${PACKAGE_NAME%.tar.gz}/${BIN}" "${PWD}" && \
-      rm -r "${PACKAGE_NAME}" && \
-      rm -r "${PACKAGE_NAME%.tar.gz}" && \
-      echo "Run './${BIN} --help' to get started"
+      echo "Extracting ${PACKAGE_NAME}..."
+      if ! tar xf "${PACKAGE_NAME}"; then
+        echo "Error: Failed to extract ${PACKAGE_NAME}"
+        exit 1
+      fi
+
+      # Find the binary in the extracted directory
+      extracted_dir="${PACKAGE_NAME%.tar.gz}"
+      if [ -f "${extracted_dir}/${BIN}" ]; then
+        mv "${extracted_dir}/${BIN}" "${PWD}/"
+        rm -f "${PACKAGE_NAME}"
+        rm -rf "${extracted_dir}"
+        chmod +x "${BIN}"
+        echo "Installation completed successfully!"
+        echo "Run './${BIN} --help' to get started"
+      else
+        echo "Error: Binary ${BIN} not found in extracted archive"
+        rm -f "${PACKAGE_NAME}"
+        rm -rf "${extracted_dir}"
+        exit 1
+      fi
     fi
   fi
 }

src/api/Cargo.toml

@@ -19,6 +19,3 @@ paste.workspace = true
 prost.workspace = true
 serde_json.workspace = true
 snafu.workspace = true
-
-[build-dependencies]
-tonic-build = "0.11"

src/api/src/error.rs

@@ -17,6 +17,7 @@ use std::any::Any;
 use common_error::ext::ErrorExt;
 use common_error::status_code::StatusCode;
 use common_macro::stack_trace_debug;
+use common_time::timestamp::TimeUnit;
 use datatypes::prelude::ConcreteDataType;
 use snafu::prelude::*;
 use snafu::Location;
@@ -66,12 +67,28 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display("Invalid time unit: {time_unit}"))]
+    InvalidTimeUnit {
+        time_unit: i32,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Inconsistent time unit: {:?}", units))]
+    InconsistentTimeUnit {
+        units: Vec<TimeUnit>,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 impl ErrorExt for Error {
     fn status_code(&self) -> StatusCode {
         match self {
-            Error::UnknownColumnDataType { .. } => StatusCode::InvalidArguments,
+            Error::UnknownColumnDataType { .. }
+            | Error::InvalidTimeUnit { .. }
+            | Error::InconsistentTimeUnit { .. } => StatusCode::InvalidArguments,
             Error::IntoColumnDataType { .. } | Error::SerializeJson { .. } => {
                 StatusCode::Unexpected
             }

src/api/src/helper.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashSet;
 use std::sync::Arc;
 
 use common_base::BitVec;
@@ -46,7 +47,7 @@ use greptime_proto::v1::{
 use paste::paste;
 use snafu::prelude::*;
 
-use crate::error::{self, Result};
+use crate::error::{self, InconsistentTimeUnitSnafu, InvalidTimeUnitSnafu, Result};
 use crate::v1::column::Values;
 use crate::v1::{Column, ColumnDataType, Value as GrpcValue};
 
@@ -291,6 +292,7 @@ impl TryFrom<ConcreteDataType> for ColumnDataTypeWrapper {
             ConcreteDataType::Vector(_) => ColumnDataType::Vector,
             ConcreteDataType::Null(_)
             | ConcreteDataType::List(_)
+            | ConcreteDataType::Struct(_)
             | ConcreteDataType::Dictionary(_)
             | ConcreteDataType::Duration(_) => {
                 return error::IntoColumnDataTypeSnafu { from: datatype }.fail()
@@ -703,6 +705,7 @@ pub fn pb_values_to_vector_ref(data_type: &ConcreteDataType, values: Values) ->
         ConcreteDataType::Vector(_) => Arc::new(BinaryVector::from_vec(values.binary_values)),
         ConcreteDataType::Null(_)
         | ConcreteDataType::List(_)
+        | ConcreteDataType::Struct(_)
         | ConcreteDataType::Dictionary(_)
         | ConcreteDataType::Duration(_)
         | ConcreteDataType::Json(_) => {
@@ -864,6 +867,7 @@ pub fn pb_values_to_values(data_type: &ConcreteDataType, values: Values) -> Vec<
         ConcreteDataType::Vector(_) => values.binary_values.into_iter().map(|v| v.into()).collect(),
         ConcreteDataType::Null(_)
         | ConcreteDataType::List(_)
+        | ConcreteDataType::Struct(_)
         | ConcreteDataType::Dictionary(_)
         | ConcreteDataType::Duration(_)
         | ConcreteDataType::Json(_) => {
@@ -1076,6 +1080,89 @@ pub fn value_to_grpc_value(value: Value) -> GrpcValue {
     }
 }
 
+pub fn from_pb_time_unit(unit: v1::TimeUnit) -> TimeUnit {
+    match unit {
+        v1::TimeUnit::Second => TimeUnit::Second,
+        v1::TimeUnit::Millisecond => TimeUnit::Millisecond,
+        v1::TimeUnit::Microsecond => TimeUnit::Microsecond,
+        v1::TimeUnit::Nanosecond => TimeUnit::Nanosecond,
+    }
+}
+
+pub fn to_pb_time_unit(unit: TimeUnit) -> v1::TimeUnit {
+    match unit {
+        TimeUnit::Second => v1::TimeUnit::Second,
+        TimeUnit::Millisecond => v1::TimeUnit::Millisecond,
+        TimeUnit::Microsecond => v1::TimeUnit::Microsecond,
+        TimeUnit::Nanosecond => v1::TimeUnit::Nanosecond,
+    }
+}
+
+pub fn from_pb_time_ranges(time_ranges: v1::TimeRanges) -> Result<Vec<(Timestamp, Timestamp)>> {
+    if time_ranges.time_ranges.is_empty() {
+        return Ok(vec![]);
+    }
+    let proto_time_unit = v1::TimeUnit::try_from(time_ranges.time_unit).map_err(|_| {
+        InvalidTimeUnitSnafu {
+            time_unit: time_ranges.time_unit,
+        }
+        .build()
+    })?;
+    let time_unit = from_pb_time_unit(proto_time_unit);
+    Ok(time_ranges
+        .time_ranges
+        .into_iter()
+        .map(|r| {
+            (
+                Timestamp::new(r.start, time_unit),
+                Timestamp::new(r.end, time_unit),
+            )
+        })
+        .collect())
+}
+
+/// All time_ranges must be of the same time unit.
+///
+/// if input `time_ranges` is empty, it will return a default `TimeRanges` with `Millisecond` as the time unit.
+pub fn to_pb_time_ranges(time_ranges: &[(Timestamp, Timestamp)]) -> Result<v1::TimeRanges> {
+    let is_same_time_unit = time_ranges.windows(2).all(|x| {
+        x[0].0.unit() == x[1].0.unit()
+            && x[0].1.unit() == x[1].1.unit()
+            && x[0].0.unit() == x[0].1.unit()
+    });
+
+    if !is_same_time_unit {
+        let all_time_units: Vec<_> = time_ranges
+            .iter()
+            .map(|(s, e)| [s.unit(), e.unit()])
+            .clone()
+            .flatten()
+            .collect::<HashSet<_>>()
+            .into_iter()
+            .collect();
+        InconsistentTimeUnitSnafu {
+            units: all_time_units,
+        }
+        .fail()?
+    }
+
+    let mut pb_time_ranges = v1::TimeRanges {
+        // default time unit is Millisecond
+        time_unit: v1::TimeUnit::Millisecond as i32,
+        time_ranges: Vec::with_capacity(time_ranges.len()),
+    };
+    if let Some((start, _end)) = time_ranges.first() {
+        pb_time_ranges.time_unit = to_pb_time_unit(start.unit()) as i32;
+    }
+    for (start, end) in time_ranges {
+        pb_time_ranges.time_ranges.push(v1::TimeRange {
+            start: start.value(),
+            end: end.value(),
+        });
+    }
+    Ok(pb_time_ranges)
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;

src/api/src/v1.rs

@@ -14,6 +14,8 @@
 
 pub mod column_def;
 
+pub mod helper;
+
 pub mod meta {
     pub use greptime_proto::v1::meta::*;
 }

src/api/src/v1/column_def.rs

@@ -24,7 +24,7 @@ use greptime_proto::v1::{
 };
 use snafu::ResultExt;
 
-use crate::error::{self, Result};
+use crate::error::{self, ConvertColumnDefaultConstraintSnafu, Result};
 use crate::helper::ColumnDataTypeWrapper;
 use crate::v1::{ColumnDef, ColumnOptions, SemanticType};
 
@@ -77,6 +77,48 @@ pub fn try_as_column_schema(column_def: &ColumnDef) -> Result<ColumnSchema> {
         })
 }
 
+/// Tries to construct a `ColumnDef` from the given `ColumnSchema`.
+///
+/// TODO(weny): Add tests for this function.
+pub fn try_as_column_def(column_schema: &ColumnSchema, is_primary_key: bool) -> Result<ColumnDef> {
+    let column_datatype =
+        ColumnDataTypeWrapper::try_from(column_schema.data_type.clone()).map(|w| w.to_parts())?;
+
+    let semantic_type = if column_schema.is_time_index() {
+        SemanticType::Timestamp
+    } else if is_primary_key {
+        SemanticType::Tag
+    } else {
+        SemanticType::Field
+    } as i32;
+    let comment = column_schema
+        .metadata()
+        .get(COMMENT_KEY)
+        .cloned()
+        .unwrap_or_default();
+
+    let default_constraint = match column_schema.default_constraint() {
+        None => vec![],
+        Some(v) => v
+            .clone()
+            .try_into()
+            .context(ConvertColumnDefaultConstraintSnafu {
+                column: &column_schema.name,
+            })?,
+    };
+    let options = options_from_column_schema(column_schema);
+    Ok(ColumnDef {
+        name: column_schema.name.clone(),
+        data_type: column_datatype.0 as i32,
+        is_nullable: column_schema.is_nullable(),
+        default_constraint,
+        semantic_type,
+        comment,
+        datatype_extension: column_datatype.1,
+        options,
+    })
+}
+
 /// Constructs a `ColumnOptions` from the given `ColumnSchema`.
 pub fn options_from_column_schema(column_schema: &ColumnSchema) -> Option<ColumnOptions> {
     let mut options = ColumnOptions::default();

src/api/src/v1/helper.rs

@@ -0,0 +1,65 @@
+// 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 greptime_proto::v1::value::ValueData;
+use greptime_proto::v1::{ColumnDataType, ColumnSchema, Row, SemanticType, Value};
+
+/// Create a time index [ColumnSchema] with column's name and datatype.
+/// Other fields are left default.
+/// Useful when you just want to create a simple [ColumnSchema] without providing much struct fields.
+pub fn time_index_column_schema(name: &str, datatype: ColumnDataType) -> ColumnSchema {
+    ColumnSchema {
+        column_name: name.to_string(),
+        datatype: datatype as i32,
+        semantic_type: SemanticType::Timestamp as i32,
+        ..Default::default()
+    }
+}
+
+/// Create a tag [ColumnSchema] with column's name and datatype.
+/// Other fields are left default.
+/// Useful when you just want to create a simple [ColumnSchema] without providing much struct fields.
+pub fn tag_column_schema(name: &str, datatype: ColumnDataType) -> ColumnSchema {
+    ColumnSchema {
+        column_name: name.to_string(),
+        datatype: datatype as i32,
+        semantic_type: SemanticType::Tag as i32,
+        ..Default::default()
+    }
+}
+
+/// Create a field [ColumnSchema] with column's name and datatype.
+/// Other fields are left default.
+/// Useful when you just want to create a simple [ColumnSchema] without providing much struct fields.
+pub fn field_column_schema(name: &str, datatype: ColumnDataType) -> ColumnSchema {
+    ColumnSchema {
+        column_name: name.to_string(),
+        datatype: datatype as i32,
+        semantic_type: SemanticType::Field as i32,
+        ..Default::default()
+    }
+}
+
+/// Create a [Row] from [ValueData]s.
+/// Useful when you don't want to write much verbose codes.
+pub fn row(values: Vec<ValueData>) -> Row {
+    Row {
+        values: values
+            .into_iter()
+            .map(|x| Value {
+                value_data: Some(x),
+            })
+            .collect::<Vec<_>>(),
+    }
+}

src/catalog/Cargo.toml

@@ -21,6 +21,7 @@ bytes.workspace = true
 common-base.workspace = true
 common-catalog.workspace = true
 common-error.workspace = true
+common-event-recorder.workspace = true
 common-frontend.workspace = true
 common-macro.workspace = true
 common-meta.workspace = true
@@ -44,6 +45,8 @@ moka = { workspace = true, features = ["future", "sync"] }
 partition.workspace = true
 paste.workspace = true
 prometheus.workspace = true
+promql-parser.workspace = true
+rand.workspace = true
 rustc-hash.workspace = true
 serde_json.workspace = true
 session.workspace = true

src/catalog/src/information_extension.rs

@@ -16,8 +16,8 @@ use api::v1::meta::ProcedureStatus;
 use common_error::ext::BoxedError;
 use common_meta::cluster::{ClusterInfo, NodeInfo};
 use common_meta::datanode::RegionStat;
-use common_meta::ddl::{ExecutorContext, ProcedureExecutor};
 use common_meta::key::flow::flow_state::FlowStat;
+use common_meta::procedure_executor::{ExecutorContext, ProcedureExecutor};
 use common_meta::rpc::procedure;
 use common_procedure::{ProcedureInfo, ProcedureState};
 use meta_client::MetaClientRef;

src/catalog/src/kvbackend/manager.rs

@@ -23,7 +23,8 @@ use common_catalog::consts::{
 };
 use common_error::ext::BoxedError;
 use common_meta::cache::{
-    LayeredCacheRegistryRef, TableRoute, TableRouteCacheRef, ViewInfoCacheRef,
+    LayeredCacheRegistryRef, TableInfoCacheRef, TableNameCacheRef, TableRoute, TableRouteCacheRef,
+    ViewInfoCacheRef,
 };
 use common_meta::key::catalog_name::CatalogNameKey;
 use common_meta::key::flow::FlowMetadataManager;
@@ -41,8 +42,9 @@ use session::context::{Channel, QueryContext};
 use snafu::prelude::*;
 use store_api::metric_engine_consts::METRIC_ENGINE_NAME;
 use table::dist_table::DistTable;
-use table::metadata::TableId;
+use table::metadata::{TableId, TableInfoRef};
 use table::table::numbers::{NumbersTable, NUMBERS_TABLE_NAME};
+use table::table::PartitionRules;
 use table::table_name::TableName;
 use table::TableRef;
 use tokio::sync::Semaphore;
@@ -131,6 +133,8 @@ impl KvBackendCatalogManager {
         {
             let mut new_table_info = (*table.table_info()).clone();
 
+            let mut phy_part_cols_not_in_logical_table = vec![];
+
             // Remap partition key indices from physical table to logical table
             new_table_info.meta.partition_key_indices = physical_table_info_value
                 .table_info
@@ -147,15 +151,30 @@ impl KvBackendCatalogManager {
                         .get(physical_index)
                         .and_then(|physical_column| {
                             // Find the corresponding index in the logical table schema
-                            new_table_info
+                            let idx = new_table_info
                                 .meta
                                 .schema
-                                .column_index_by_name(physical_column.name.as_str())
+                                .column_index_by_name(physical_column.name.as_str());
+                            if idx.is_none() {
+                                // not all part columns in physical table that are also in logical table
+                                phy_part_cols_not_in_logical_table
+                                    .push(physical_column.name.clone());
+                            }
+
+                            idx
                         })
                 })
                 .collect();
 
-            let new_table = DistTable::table(Arc::new(new_table_info));
+            let partition_rules = if !phy_part_cols_not_in_logical_table.is_empty() {
+                Some(PartitionRules {
+                    extra_phy_cols_not_in_logical_table: phy_part_cols_not_in_logical_table,
+                })
+            } else {
+                None
+            };
+
+            let new_table = DistTable::table_partitioned(Arc::new(new_table_info), partition_rules);
 
             return Ok(new_table);
         }
@@ -325,6 +344,63 @@ impl CatalogManager for KvBackendCatalogManager {
         Ok(None)
     }
 
+    async fn table_id(
+        &self,
+        catalog_name: &str,
+        schema_name: &str,
+        table_name: &str,
+        query_ctx: Option<&QueryContext>,
+    ) -> Result<Option<TableId>> {
+        let channel = query_ctx.map_or(Channel::Unknown, |ctx| ctx.channel());
+        if let Some(table) =
+            self.system_catalog
+                .table(catalog_name, schema_name, table_name, query_ctx)
+        {
+            return Ok(Some(table.table_info().table_id()));
+        }
+
+        let table_cache: TableNameCacheRef =
+            self.cache_registry.get().context(CacheNotFoundSnafu {
+                name: "table_name_cache",
+            })?;
+
+        let table = table_cache
+            .get_by_ref(&TableName {
+                catalog_name: catalog_name.to_string(),
+                schema_name: schema_name.to_string(),
+                table_name: table_name.to_string(),
+            })
+            .await
+            .context(GetTableCacheSnafu)?;
+
+        if let Some(table) = table {
+            return Ok(Some(table));
+        }
+
+        if channel == Channel::Postgres {
+            // falldown to pg_catalog
+            if let Some(table) =
+                self.system_catalog
+                    .table(catalog_name, PG_CATALOG_NAME, table_name, query_ctx)
+            {
+                return Ok(Some(table.table_info().table_id()));
+            }
+        }
+
+        Ok(None)
+    }
+
+    async fn table_info_by_id(&self, table_id: TableId) -> Result<Option<TableInfoRef>> {
+        let table_info_cache: TableInfoCacheRef =
+            self.cache_registry.get().context(CacheNotFoundSnafu {
+                name: "table_info_cache",
+            })?;
+        table_info_cache
+            .get_by_ref(&table_id)
+            .await
+            .context(GetTableCacheSnafu)
+    }
+
     async fn tables_by_ids(
         &self,
         catalog: &str,

src/catalog/src/lib.rs

@@ -25,7 +25,7 @@ use common_catalog::consts::{INFORMATION_SCHEMA_NAME, PG_CATALOG_NAME};
 use futures::future::BoxFuture;
 use futures_util::stream::BoxStream;
 use session::context::QueryContext;
-use table::metadata::TableId;
+use table::metadata::{TableId, TableInfoRef};
 use table::TableRef;
 
 use crate::error::Result;
@@ -89,6 +89,23 @@ pub trait CatalogManager: Send + Sync {
         query_ctx: Option<&QueryContext>,
     ) -> Result<Option<TableRef>>;
 
+    /// Returns the table id of provided table ident.
+    async fn table_id(
+        &self,
+        catalog: &str,
+        schema: &str,
+        table_name: &str,
+        query_ctx: Option<&QueryContext>,
+    ) -> Result<Option<TableId>> {
+        Ok(self
+            .table(catalog, schema, table_name, query_ctx)
+            .await?
+            .map(|t| t.table_info().ident.table_id))
+    }
+
+    /// Returns the table of provided id.
+    async fn table_info_by_id(&self, table_id: TableId) -> Result<Option<TableInfoRef>>;
+
     /// Returns the tables by table ids.
     async fn tables_by_ids(
         &self,

src/catalog/src/memory/manager.rs

@@ -28,7 +28,7 @@ use common_meta::kv_backend::memory::MemoryKvBackend;
 use futures_util::stream::BoxStream;
 use session::context::QueryContext;
 use snafu::OptionExt;
-use table::metadata::TableId;
+use table::metadata::{TableId, TableInfoRef};
 use table::TableRef;
 
 use crate::error::{CatalogNotFoundSnafu, Result, SchemaNotFoundSnafu, TableExistsSnafu};
@@ -38,7 +38,7 @@ use crate::{CatalogManager, DeregisterTableRequest, RegisterSchemaRequest, Regis
 
 type SchemaEntries = HashMap<String, HashMap<String, TableRef>>;
 
-/// Simple in-memory list of catalogs
+/// Simple in-memory list of catalogs used for tests.
 #[derive(Clone)]
 pub struct MemoryCatalogManager {
     /// Collection of catalogs containing schemas and ultimately Tables
@@ -144,6 +144,18 @@ impl CatalogManager for MemoryCatalogManager {
         Ok(result)
     }
 
+    async fn table_info_by_id(&self, table_id: TableId) -> Result<Option<TableInfoRef>> {
+        Ok(self
+            .catalogs
+            .read()
+            .unwrap()
+            .iter()
+            .flat_map(|(_, schema_entries)| schema_entries.values())
+            .flat_map(|tables| tables.values())
+            .find(|t| t.table_info().ident.table_id == table_id)
+            .map(|t| t.table_info()))
+    }
+
     async fn tables_by_ids(
         &self,
         catalog: &str,

src/catalog/src/process_manager.rs

@@ -14,17 +14,24 @@
 
 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
-use std::fmt::{Debug, Formatter};
+use std::fmt::{Debug, Display, Formatter};
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::{Arc, RwLock};
+use std::time::{Duration, Instant, UNIX_EPOCH};
 
 use api::v1::frontend::{KillProcessRequest, ListProcessRequest, ProcessInfo};
 use common_base::cancellation::CancellationHandle;
+use common_event_recorder::EventRecorderRef;
 use common_frontend::selector::{FrontendSelector, MetaClientSelector};
-use common_telemetry::{debug, info, warn};
+use common_frontend::slow_query_event::SlowQueryEvent;
+use common_telemetry::logging::SlowQueriesRecordType;
+use common_telemetry::{debug, info, slow, warn};
 use common_time::util::current_time_millis;
 use meta_client::MetaClientRef;
+use promql_parser::parser::EvalStmt;
+use rand::random;
 use snafu::{ensure, OptionExt, ResultExt};
+use sql::statements::statement::Statement;
 
 use crate::error;
 use crate::metrics::{PROCESS_KILL_COUNT, PROCESS_LIST_COUNT};
@@ -44,6 +51,23 @@ pub struct ProcessManager {
     frontend_selector: Option<MetaClientSelector>,
 }
 
+/// Represents a parsed query statement, functionally equivalent to [query::parser::QueryStatement].
+/// This enum is defined here to avoid cyclic dependencies with the query parser module.
+#[derive(Debug, Clone)]
+pub enum QueryStatement {
+    Sql(Statement),
+    Promql(EvalStmt),
+}
+
+impl Display for QueryStatement {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            QueryStatement::Sql(stmt) => write!(f, "{}", stmt),
+            QueryStatement::Promql(eval_stmt) => write!(f, "{}", eval_stmt),
+        }
+    }
+}
+
 impl ProcessManager {
     /// Create a [ProcessManager] instance with server address and kv client.
     pub fn new(server_addr: String, meta_client: Option<MetaClientRef>) -> Self {
@@ -67,6 +91,7 @@ impl ProcessManager {
         query: String,
         client: String,
         query_id: Option<ProcessId>,
+        _slow_query_timer: Option<SlowQueryTimer>,
     ) -> Ticket {
         let id = query_id.unwrap_or_else(|| self.next_id.fetch_add(1, Ordering::Relaxed));
         let process = ProcessInfo {
@@ -93,6 +118,7 @@ impl ProcessManager {
             manager: self.clone(),
             id,
             cancellation_handle,
+            _slow_query_timer,
         }
     }
 
@@ -223,6 +249,9 @@ pub struct Ticket {
     pub(crate) manager: ProcessManagerRef,
     pub(crate) id: ProcessId,
     pub cancellation_handle: Arc<CancellationHandle>,
+
+    // Keep the handle of the slow query timer to ensure it will trigger the event recording when dropped.
+    _slow_query_timer: Option<SlowQueryTimer>,
 }
 
 impl Drop for Ticket {
@@ -263,6 +292,114 @@ impl Debug for CancellableProcess {
     }
 }
 
+/// SlowQueryTimer is used to log slow query when it's dropped.
+/// In drop(), it will check if the query is slow and send the slow query event to the handler.
+pub struct SlowQueryTimer {
+    start: Instant,
+    stmt: QueryStatement,
+    threshold: Duration,
+    sample_ratio: f64,
+    record_type: SlowQueriesRecordType,
+    recorder: EventRecorderRef,
+}
+
+impl SlowQueryTimer {
+    pub fn new(
+        stmt: QueryStatement,
+        threshold: Duration,
+        sample_ratio: f64,
+        record_type: SlowQueriesRecordType,
+        recorder: EventRecorderRef,
+    ) -> Self {
+        Self {
+            start: Instant::now(),
+            stmt,
+            threshold,
+            sample_ratio,
+            record_type,
+            recorder,
+        }
+    }
+}
+
+impl SlowQueryTimer {
+    fn send_slow_query_event(&self, elapsed: Duration) {
+        let mut slow_query_event = SlowQueryEvent {
+            cost: elapsed.as_millis() as u64,
+            threshold: self.threshold.as_millis() as u64,
+            query: "".to_string(),
+
+            // The following fields are only used for PromQL queries.
+            is_promql: false,
+            promql_range: None,
+            promql_step: None,
+            promql_start: None,
+            promql_end: None,
+        };
+
+        match &self.stmt {
+            QueryStatement::Promql(stmt) => {
+                slow_query_event.is_promql = true;
+                slow_query_event.query = stmt.expr.to_string();
+                slow_query_event.promql_step = Some(stmt.interval.as_millis() as u64);
+
+                let start = stmt
+                    .start
+                    .duration_since(UNIX_EPOCH)
+                    .unwrap_or_default()
+                    .as_millis() as i64;
+
+                let end = stmt
+                    .end
+                    .duration_since(UNIX_EPOCH)
+                    .unwrap_or_default()
+                    .as_millis() as i64;
+
+                slow_query_event.promql_range = Some((end - start) as u64);
+                slow_query_event.promql_start = Some(start);
+                slow_query_event.promql_end = Some(end);
+            }
+            QueryStatement::Sql(stmt) => {
+                slow_query_event.query = stmt.to_string();
+            }
+        }
+
+        match self.record_type {
+            // Send the slow query event to the event recorder to persist it as the system table.
+            SlowQueriesRecordType::SystemTable => {
+                self.recorder.record(Box::new(slow_query_event));
+            }
+            // Record the slow query in a specific logs file.
+            SlowQueriesRecordType::Log => {
+                slow!(
+                    cost = slow_query_event.cost,
+                    threshold = slow_query_event.threshold,
+                    query = slow_query_event.query,
+                    is_promql = slow_query_event.is_promql,
+                    promql_range = slow_query_event.promql_range,
+                    promql_step = slow_query_event.promql_step,
+                    promql_start = slow_query_event.promql_start,
+                    promql_end = slow_query_event.promql_end,
+                );
+            }
+        }
+    }
+}
+
+impl Drop for SlowQueryTimer {
+    fn drop(&mut self) {
+        // Calculate the elaspsed duration since the timer is created.
+        let elapsed = self.start.elapsed();
+        if elapsed > self.threshold {
+            // Only capture a portion of slow queries based on sample_ratio.
+            // Generate a random number in [0, 1) and compare it with sample_ratio.
+            if self.sample_ratio >= 1.0 || random::<f64>() <= self.sample_ratio {
+                self.send_slow_query_event(elapsed);
+            }
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;
@@ -278,6 +415,7 @@ mod tests {
             "SELECT * FROM table".to_string(),
             "".to_string(),
             None,
+            None,
         );
 
         let running_processes = process_manager.local_processes(None).unwrap();
@@ -301,6 +439,7 @@ mod tests {
             "SELECT * FROM table".to_string(),
             "client1".to_string(),
             Some(custom_id),
+            None,
         );
 
         assert_eq!(ticket.id, custom_id);
@@ -321,6 +460,7 @@ mod tests {
             "SELECT * FROM table1".to_string(),
             "client1".to_string(),
             None,
+            None,
         );
 
         let ticket2 = process_manager.clone().register_query(
@@ -329,6 +469,7 @@ mod tests {
             "SELECT * FROM table2".to_string(),
             "client2".to_string(),
             None,
+            None,
         );
 
         let running_processes = process_manager.local_processes(Some("public")).unwrap();
@@ -350,6 +491,7 @@ mod tests {
             "SELECT * FROM table1".to_string(),
             "client1".to_string(),
             None,
+            None,
         );
 
         let _ticket2 = process_manager.clone().register_query(
@@ -358,6 +500,7 @@ mod tests {
             "SELECT * FROM table2".to_string(),
             "client2".to_string(),
             None,
+            None,
         );
 
         // Test listing processes for specific catalog
@@ -384,6 +527,7 @@ mod tests {
             "SELECT * FROM table".to_string(),
             "client1".to_string(),
             None,
+            None,
         );
         assert_eq!(process_manager.local_processes(None).unwrap().len(), 1);
         process_manager.deregister_query("public".to_string(), ticket.id);
@@ -400,6 +544,7 @@ mod tests {
             "SELECT * FROM table".to_string(),
             "client1".to_string(),
             None,
+            None,
         );
 
         assert!(!ticket.cancellation_handle.is_cancelled());
@@ -417,6 +562,7 @@ mod tests {
             "SELECT * FROM table".to_string(),
             "client1".to_string(),
             None,
+            None,
         );
         assert!(!ticket.cancellation_handle.is_cancelled());
         let killed = process_manager
@@ -462,6 +608,7 @@ mod tests {
             "SELECT COUNT(*) FROM users WHERE age > 18".to_string(),
             "test_client".to_string(),
             Some(42),
+            None,
         );
 
         let processes = process_manager.local_processes(None).unwrap();
@@ -488,6 +635,7 @@ mod tests {
                 "SELECT * FROM table".to_string(),
                 "client1".to_string(),
                 None,
+                None,
             );
 
             // Process should be registered

src/catalog/src/system_schema/information_schema.rs

@@ -162,6 +162,16 @@ impl SystemSchemaProviderInner for InformationSchemaProvider {
     }
 
     fn system_table(&self, name: &str) -> Option<SystemTableRef> {
+        #[cfg(feature = "enterprise")]
+        if let Some(factory) = self.extra_table_factories.get(name) {
+            let req = MakeInformationTableRequest {
+                catalog_name: self.catalog_name.clone(),
+                catalog_manager: self.catalog_manager.clone(),
+                kv_backend: self.kv_backend.clone(),
+            };
+            return Some(factory.make_information_table(req));
+        }
+
         match name.to_ascii_lowercase().as_str() {
             TABLES => Some(Arc::new(InformationSchemaTables::new(
                 self.catalog_name.clone(),
@@ -240,22 +250,7 @@ impl SystemSchemaProviderInner for InformationSchemaProvider {
                 .process_manager
                 .as_ref()
                 .map(|p| Arc::new(InformationSchemaProcessList::new(p.clone())) as _),
-            table_name => {
-                #[cfg(feature = "enterprise")]
-                return self.extra_table_factories.get(table_name).map(|factory| {
-                    let req = MakeInformationTableRequest {
-                        catalog_name: self.catalog_name.clone(),
-                        catalog_manager: self.catalog_manager.clone(),
-                        kv_backend: self.kv_backend.clone(),
-                    };
-                    factory.make_information_table(req)
-                });
-                #[cfg(not(feature = "enterprise"))]
-                {
-                    let _ = table_name;
-                    None
-                }
-            }
+            _ => None,
         }
     }
 }

src/catalog/src/system_schema/information_schema/information_memory_table.rs

@@ -15,7 +15,8 @@
 use std::sync::Arc;
 
 use common_catalog::consts::{METRIC_ENGINE, MITO_ENGINE};
-use datatypes::schema::{Schema, SchemaRef};
+use datatypes::data_type::ConcreteDataType;
+use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
 use datatypes::vectors::{Int64Vector, StringVector, VectorRef};
 
 use crate::system_schema::information_schema::table_names::*;
@@ -367,28 +368,18 @@ pub(super) fn get_schema_columns(table_name: &str) -> (SchemaRef, Vec<VectorRef>
 
         TRIGGERS => (
             vec![
-                string_column("TRIGGER_CATALOG"),
-                string_column("TRIGGER_SCHEMA"),
                 string_column("TRIGGER_NAME"),
-                string_column("EVENT_MANIPULATION"),
-                string_column("EVENT_OBJECT_CATALOG"),
-                string_column("EVENT_OBJECT_SCHEMA"),
-                string_column("EVENT_OBJECT_TABLE"),
-                bigint_column("ACTION_ORDER"),
-                string_column("ACTION_CONDITION"),
-                string_column("ACTION_STATEMENT"),
-                string_column("ACTION_ORIENTATION"),
-                string_column("ACTION_TIMING"),
-                string_column("ACTION_REFERENCE_OLD_TABLE"),
-                string_column("ACTION_REFERENCE_NEW_TABLE"),
-                string_column("ACTION_REFERENCE_OLD_ROW"),
-                string_column("ACTION_REFERENCE_NEW_ROW"),
-                timestamp_micro_column("CREATED"),
-                string_column("SQL_MODE"),
-                string_column("DEFINER"),
-                string_column("CHARACTER_SET_CLIENT"),
-                string_column("COLLATION_CONNECTION"),
-                string_column("DATABASE_COLLATION"),
+                ColumnSchema::new(
+                    "trigger_id",
+                    ConcreteDataType::uint64_datatype(),
+                    false,
+                ),
+                string_column("TRIGGER_DEFINITION"),
+                ColumnSchema::new(
+                    "flownode_id",
+                    ConcreteDataType::uint64_datatype(),
+                    true,
+                ),
             ],
             vec![],
         ),

src/catalog/src/system_schema/information_schema/partitions.rs

@@ -329,13 +329,8 @@ impl InformationSchemaPartitionsBuilder {
             self.partition_names.push(Some(&partition_name));
             self.partition_ordinal_positions
                 .push(Some((index + 1) as i64));
-            let expressions = if partition.partition.partition_columns().is_empty() {
-                None
-            } else {
-                Some(partition.partition.to_string())
-            };
-
-            self.partition_expressions.push(expressions.as_deref());
+            let expression = partition.partition_expr.as_ref().map(|e| e.to_string());
+            self.partition_expressions.push(expression.as_deref());
             self.create_times.push(Some(TimestampMicrosecond::from(
                 table_info.meta.created_on.timestamp_millis(),
             )));

src/catalog/src/system_schema/information_schema/region_statistics.rs

@@ -44,6 +44,7 @@ const DISK_SIZE: &str = "disk_size";
 const MEMTABLE_SIZE: &str = "memtable_size";
 const MANIFEST_SIZE: &str = "manifest_size";
 const SST_SIZE: &str = "sst_size";
+const SST_NUM: &str = "sst_num";
 const INDEX_SIZE: &str = "index_size";
 const ENGINE: &str = "engine";
 const REGION_ROLE: &str = "region_role";
@@ -87,6 +88,7 @@ impl InformationSchemaRegionStatistics {
             ColumnSchema::new(MEMTABLE_SIZE, ConcreteDataType::uint64_datatype(), true),
             ColumnSchema::new(MANIFEST_SIZE, ConcreteDataType::uint64_datatype(), true),
             ColumnSchema::new(SST_SIZE, ConcreteDataType::uint64_datatype(), true),
+            ColumnSchema::new(SST_NUM, ConcreteDataType::uint64_datatype(), true),
             ColumnSchema::new(INDEX_SIZE, ConcreteDataType::uint64_datatype(), true),
             ColumnSchema::new(ENGINE, ConcreteDataType::string_datatype(), true),
             ColumnSchema::new(REGION_ROLE, ConcreteDataType::string_datatype(), true),
@@ -149,6 +151,7 @@ struct InformationSchemaRegionStatisticsBuilder {
     memtable_sizes: UInt64VectorBuilder,
     manifest_sizes: UInt64VectorBuilder,
     sst_sizes: UInt64VectorBuilder,
+    sst_nums: UInt64VectorBuilder,
     index_sizes: UInt64VectorBuilder,
     engines: StringVectorBuilder,
     region_roles: StringVectorBuilder,
@@ -167,6 +170,7 @@ impl InformationSchemaRegionStatisticsBuilder {
             memtable_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
             manifest_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
             sst_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
+            sst_nums: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
             index_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
             engines: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             region_roles: StringVectorBuilder::with_capacity(INIT_CAPACITY),
@@ -197,6 +201,7 @@ impl InformationSchemaRegionStatisticsBuilder {
             (MEMTABLE_SIZE, &Value::from(region_stat.memtable_size)),
             (MANIFEST_SIZE, &Value::from(region_stat.manifest_size)),
             (SST_SIZE, &Value::from(region_stat.sst_size)),
+            (SST_NUM, &Value::from(region_stat.sst_num)),
             (INDEX_SIZE, &Value::from(region_stat.index_size)),
             (ENGINE, &Value::from(region_stat.engine.as_str())),
             (REGION_ROLE, &Value::from(region_stat.role.to_string())),
@@ -215,6 +220,7 @@ impl InformationSchemaRegionStatisticsBuilder {
         self.memtable_sizes.push(Some(region_stat.memtable_size));
         self.manifest_sizes.push(Some(region_stat.manifest_size));
         self.sst_sizes.push(Some(region_stat.sst_size));
+        self.sst_nums.push(Some(region_stat.sst_num));
         self.index_sizes.push(Some(region_stat.index_size));
         self.engines.push(Some(&region_stat.engine));
         self.region_roles.push(Some(&region_stat.role.to_string()));
@@ -230,6 +236,7 @@ impl InformationSchemaRegionStatisticsBuilder {
             Arc::new(self.memtable_sizes.finish()),
             Arc::new(self.manifest_sizes.finish()),
             Arc::new(self.sst_sizes.finish()),
+            Arc::new(self.sst_nums.finish()),
             Arc::new(self.index_sizes.finish()),
             Arc::new(self.engines.finish()),
             Arc::new(self.region_roles.finish()),

src/catalog/src/system_schema/information_schema/table_names.rs

@@ -48,4 +48,3 @@ pub const FLOWS: &str = "flows";
 pub const PROCEDURE_INFO: &str = "procedure_info";
 pub const REGION_STATISTICS: &str = "region_statistics";
 pub const PROCESS_LIST: &str = "process_list";
-pub const TRIGGER_LIST: &str = "trigger_list";

src/catalog/src/system_schema/pg_catalog/pg_class.rs

@@ -169,7 +169,7 @@ impl DfPartitionStream for PGClass {
 }
 
 /// Builds the `pg_catalog.pg_class` table row by row
-/// TODO(J0HN50N133): `relowner` is always the [`DUMMY_OWNER_ID`] cuz we don't have user.
+/// TODO(J0HN50N133): `relowner` is always the [`DUMMY_OWNER_ID`] because we don't have users.
 /// Once we have user system, make it the actual owner of the table.
 struct PGClassBuilder {
     schema: SchemaRef,

src/catalog/src/system_schema/predicate.rs

@@ -133,7 +133,7 @@ impl Predicate {
                 let Expr::Column(c) = *expr else {
                     unreachable!();
                 };
-                let Expr::Literal(ScalarValue::Utf8(Some(pattern))) = *pattern else {
+                let Expr::Literal(ScalarValue::Utf8(Some(pattern)), _) = *pattern else {
                     unreachable!();
                 };
 
@@ -148,8 +148,8 @@ impl Predicate {
             // left OP right
             Expr::BinaryExpr(bin) => match (*bin.left, bin.op, *bin.right) {
                 // left == right
-                (Expr::Literal(scalar), Operator::Eq, Expr::Column(c))
-                | (Expr::Column(c), Operator::Eq, Expr::Literal(scalar)) => {
+                (Expr::Literal(scalar, _), Operator::Eq, Expr::Column(c))
+                | (Expr::Column(c), Operator::Eq, Expr::Literal(scalar, _)) => {
                     let Ok(v) = Value::try_from(scalar) else {
                         return None;
                     };
@@ -157,8 +157,8 @@ impl Predicate {
                     Some(Predicate::Eq(c.name, v))
                 }
                 // left != right
-                (Expr::Literal(scalar), Operator::NotEq, Expr::Column(c))
-                | (Expr::Column(c), Operator::NotEq, Expr::Literal(scalar)) => {
+                (Expr::Literal(scalar, _), Operator::NotEq, Expr::Column(c))
+                | (Expr::Column(c), Operator::NotEq, Expr::Literal(scalar, _)) => {
                     let Ok(v) = Value::try_from(scalar) else {
                         return None;
                     };
@@ -189,7 +189,7 @@ impl Predicate {
                         let mut values = Vec::with_capacity(list.len());
                         for scalar in list {
                             // Safety: checked by `is_all_scalars`
-                            let Expr::Literal(scalar) = scalar else {
+                            let Expr::Literal(scalar, _) = scalar else {
                                 unreachable!();
                             };
 
@@ -237,7 +237,7 @@ fn like_utf8(s: &str, pattern: &str, case_insensitive: &bool) -> Option<bool> {
 }
 
 fn is_string_literal(expr: &Expr) -> bool {
-    matches!(expr, Expr::Literal(ScalarValue::Utf8(Some(_))))
+    matches!(expr, Expr::Literal(ScalarValue::Utf8(Some(_)), _))
 }
 
 fn is_column(expr: &Expr) -> bool {
@@ -286,14 +286,14 @@ impl Predicates {
 
 /// Returns true when the values are all [`DfExpr::Literal`].
 fn is_all_scalars(list: &[Expr]) -> bool {
-    list.iter().all(|v| matches!(v, Expr::Literal(_)))
+    list.iter().all(|v| matches!(v, Expr::Literal(_, _)))
 }
 
 #[cfg(test)]
 mod tests {
-    use datafusion::common::{Column, ScalarValue};
+    use datafusion::common::Column;
     use datafusion::logical_expr::expr::InList;
-    use datafusion::logical_expr::BinaryExpr;
+    use datafusion::logical_expr::{BinaryExpr, Literal};
 
     use super::*;
 
@@ -378,7 +378,7 @@ mod tests {
         let expr = Expr::Like(Like {
             negated: false,
             expr: Box::new(column("a")),
-            pattern: Box::new(string_literal("%abc")),
+            pattern: Box::new("%abc".lit()),
             case_insensitive: true,
             escape_char: None,
         });
@@ -405,7 +405,7 @@ mod tests {
         let expr = Expr::Like(Like {
             negated: false,
             expr: Box::new(column("a")),
-            pattern: Box::new(string_literal("%abc")),
+            pattern: Box::new("%abc".lit()),
             case_insensitive: false,
             escape_char: None,
         });
@@ -425,7 +425,7 @@ mod tests {
         let expr = Expr::Like(Like {
             negated: true,
             expr: Box::new(column("a")),
-            pattern: Box::new(string_literal("%abc")),
+            pattern: Box::new("%abc".lit()),
             case_insensitive: true,
             escape_char: None,
         });
@@ -440,10 +440,6 @@ mod tests {
         Expr::Column(Column::from_name(name))
     }
 
-    fn string_literal(v: &str) -> Expr {
-        Expr::Literal(ScalarValue::Utf8(Some(v.to_string())))
-    }
-
     fn match_string_value(v: &Value, expected: &str) -> bool {
         matches!(v, Value::String(bs) if bs.as_utf8() == expected)
     }
@@ -463,13 +459,13 @@ mod tests {
         let expr1 = Expr::BinaryExpr(BinaryExpr {
             left: Box::new(column("a")),
             op: Operator::Eq,
-            right: Box::new(string_literal("a_value")),
+            right: Box::new("a_value".lit()),
         });
 
         let expr2 = Expr::BinaryExpr(BinaryExpr {
             left: Box::new(column("b")),
             op: Operator::NotEq,
-            right: Box::new(string_literal("b_value")),
+            right: Box::new("b_value".lit()),
         });
 
         (expr1, expr2)
@@ -508,7 +504,7 @@ mod tests {
 
         let inlist_expr = Expr::InList(InList {
             expr: Box::new(column("a")),
-            list: vec![string_literal("a1"), string_literal("a2")],
+            list: vec!["a1".lit(), "a2".lit()],
             negated: false,
         });
 
@@ -518,7 +514,7 @@ mod tests {
 
         let inlist_expr = Expr::InList(InList {
             expr: Box::new(column("a")),
-            list: vec![string_literal("a1"), string_literal("a2")],
+            list: vec!["a1".lit(), "a2".lit()],
             negated: true,
         });
         let inlist_p = Predicate::from_expr(inlist_expr).unwrap();

src/catalog/src/table_source.rs

@@ -32,7 +32,7 @@ use dummy_catalog::DummyCatalogList;
 use table::TableRef;
 
 use crate::error::{
-    CastManagerSnafu, DatafusionSnafu, DecodePlanSnafu, GetViewCacheSnafu, ProjectViewColumnsSnafu,
+    CastManagerSnafu, DecodePlanSnafu, GetViewCacheSnafu, ProjectViewColumnsSnafu,
     QueryAccessDeniedSnafu, Result, TableNotExistSnafu, ViewInfoNotFoundSnafu,
     ViewPlanColumnsChangedSnafu,
 };
@@ -70,6 +70,11 @@ impl DfTableSourceProvider {
         }
     }
 
+    /// Returns the query context.
+    pub fn query_ctx(&self) -> &QueryContextRef {
+        &self.query_ctx
+    }
+
     pub fn resolve_table_ref(&self, table_ref: TableReference) -> Result<ResolvedTableReference> {
         if self.disallow_cross_catalog_query {
             match &table_ref {
@@ -194,10 +199,10 @@ impl DfTableSourceProvider {
             logical_plan
         };
 
-        Ok(Arc::new(
-            ViewTable::try_new(logical_plan, Some(view_info.definition.to_string()))
-                .context(DatafusionSnafu)?,
-        ))
+        Ok(Arc::new(ViewTable::new(
+            logical_plan,
+            Some(view_info.definition.to_string()),
+        )))
     }
 }
 

src/cli/src/bench.rs

@@ -188,6 +188,7 @@ fn create_region_routes(regions: Vec<RegionNumber>) -> Vec<RegionRoute> {
                 name: String::new(),
                 partition: None,
                 attrs: BTreeMap::new(),
+                partition_expr: Default::default(),
             },
             leader_peer: Some(Peer {
                 id: rng.random_range(0..10),

src/cli/src/metadata/common.rs

@@ -75,7 +75,7 @@ impl StoreConfig {
                 #[cfg(feature = "pg_kvbackend")]
                 BackendImpl::PostgresStore => {
                     let table_name = &self.meta_table_name;
-                    let pool = meta_srv::bootstrap::create_postgres_pool(store_addrs)
+                    let pool = meta_srv::bootstrap::create_postgres_pool(store_addrs, None)
                         .await
                         .map_err(BoxedError::new)?;
                     Ok(common_meta::kv_backend::rds::PgStore::with_pg_pool(

src/cli/src/metadata/repair/create_table.rs

@@ -74,11 +74,19 @@ pub fn make_create_region_request_for_peer(
     let catalog = &create_table_expr.catalog_name;
     let schema = &create_table_expr.schema_name;
     let storage_path = region_storage_path(catalog, schema);
+    let partition_exprs = region_routes
+        .iter()
+        .map(|r| (r.region.id.region_number(), r.region.partition_expr()))
+        .collect::<HashMap<_, _>>();
 
     for region_number in &regions_on_this_peer {
         let region_id = RegionId::new(logical_table_id, *region_number);
-        let region_request =
-            request_builder.build_one(region_id, storage_path.clone(), &HashMap::new());
+        let region_request = request_builder.build_one(
+            region_id,
+            storage_path.clone(),
+            &HashMap::new(),
+            &partition_exprs,
+        );
         requests.push(region_request);
     }
 

src/client/Cargo.toml

@@ -29,6 +29,7 @@ datatypes.workspace = true
 enum_dispatch = "0.3"
 futures.workspace = true
 futures-util.workspace = true
+humantime.workspace = true
 lazy_static.workspace = true
 moka = { workspace = true, features = ["future"] }
 parking_lot.workspace = true
@@ -38,6 +39,7 @@ query.workspace = true
 rand.workspace = true
 serde_json.workspace = true
 snafu.workspace = true
+store-api.workspace = true
 substrait.workspace = true
 tokio.workspace = true
 tokio-stream = { workspace = true, features = ["net"] }

src/client/src/client_manager.rs

@@ -17,7 +17,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use common_grpc::channel_manager::{ChannelConfig, ChannelManager};
-use common_meta::node_manager::{DatanodeRef, FlownodeRef, NodeManager};
+use common_meta::node_manager::{DatanodeManager, DatanodeRef, FlownodeManager, FlownodeRef};
 use common_meta::peer::Peer;
 use moka::future::{Cache, CacheBuilder};
 
@@ -45,7 +45,7 @@ impl Debug for NodeClients {
 }
 
 #[async_trait::async_trait]
-impl NodeManager for NodeClients {
+impl DatanodeManager for NodeClients {
     async fn datanode(&self, datanode: &Peer) -> DatanodeRef {
         let client = self.get_client(datanode).await;
 
@@ -60,7 +60,10 @@ impl NodeManager for NodeClients {
             *accept_compression,
         ))
     }
+}
 
+#[async_trait::async_trait]
+impl FlownodeManager for NodeClients {
     async fn flownode(&self, flownode: &Peer) -> FlownodeRef {
         let client = self.get_client(flownode).await;
 

src/client/src/database.rs

@@ -23,7 +23,7 @@ use api::v1::greptime_request::Request;
 use api::v1::query_request::Query;
 use api::v1::{
     AlterTableExpr, AuthHeader, Basic, CreateTableExpr, DdlRequest, GreptimeRequest,
-    InsertRequests, QueryRequest, RequestHeader,
+    InsertRequests, QueryRequest, RequestHeader, RowInsertRequests,
 };
 use arrow_flight::{FlightData, Ticket};
 use async_stream::stream;
@@ -42,7 +42,7 @@ use common_telemetry::{error, warn};
 use futures::future;
 use futures_util::{Stream, StreamExt, TryStreamExt};
 use prost::Message;
-use snafu::{ensure, ResultExt};
+use snafu::{ensure, OptionExt, ResultExt};
 use tonic::metadata::{AsciiMetadataKey, AsciiMetadataValue, MetadataMap, MetadataValue};
 use tonic::transport::Channel;
 
@@ -75,12 +75,24 @@ pub struct Database {
 }
 
 pub struct DatabaseClient {
+    pub addr: String,
     pub inner: GreptimeDatabaseClient<Channel>,
 }
 
+impl DatabaseClient {
+    /// Returns a closure that logs the error when the request fails.
+    pub fn inspect_err<'a>(&'a self, context: &'a str) -> impl Fn(&tonic::Status) + 'a {
+        let addr = &self.addr;
+        move |status| {
+            error!("Failed to {context} request, peer: {addr}, status: {status:?}");
+        }
+    }
+}
+
 fn make_database_client(client: &Client) -> Result<DatabaseClient> {
-    let (_, channel) = client.find_channel()?;
+    let (addr, channel) = client.find_channel()?;
     Ok(DatabaseClient {
+        addr,
         inner: GreptimeDatabaseClient::new(channel)
             .max_decoding_message_size(client.max_grpc_recv_message_size())
             .max_encoding_message_size(client.max_grpc_send_message_size()),
@@ -118,6 +130,7 @@ impl Database {
         }
     }
 
+    /// Set the catalog for the database client.
     pub fn set_catalog(&mut self, catalog: impl Into<String>) {
         self.catalog = catalog.into();
     }
@@ -130,6 +143,7 @@ impl Database {
         }
     }
 
+    /// Set the schema for the database client.
     pub fn set_schema(&mut self, schema: impl Into<String>) {
         self.schema = schema.into();
     }
@@ -142,33 +156,69 @@ impl Database {
         }
     }
 
+    /// Set the timezone for the database client.
     pub fn set_timezone(&mut self, timezone: impl Into<String>) {
         self.timezone = timezone.into();
     }
 
+    /// Set the auth scheme for the database client.
     pub fn set_auth(&mut self, auth: AuthScheme) {
         self.ctx.auth_header = Some(AuthHeader {
             auth_scheme: Some(auth),
         });
     }
 
+    /// Make an InsertRequests request to the database.
     pub async fn insert(&self, requests: InsertRequests) -> Result<u32> {
         self.handle(Request::Inserts(requests)).await
     }
 
+    /// Make an InsertRequests request to the database with hints.
     pub async fn insert_with_hints(
         &self,
         requests: InsertRequests,
         hints: &[(&str, &str)],
     ) -> Result<u32> {
-        let mut client = make_database_client(&self.client)?.inner;
+        let mut client = make_database_client(&self.client)?;
         let request = self.to_rpc_request(Request::Inserts(requests));
 
         let mut request = tonic::Request::new(request);
         let metadata = request.metadata_mut();
         Self::put_hints(metadata, hints)?;
 
-        let response = client.handle(request).await?.into_inner();
+        let response = client
+            .inner
+            .handle(request)
+            .await
+            .inspect_err(client.inspect_err("insert_with_hints"))?
+            .into_inner();
+        from_grpc_response(response)
+    }
+
+    /// Make a RowInsertRequests request to the database.
+    pub async fn row_inserts(&self, requests: RowInsertRequests) -> Result<u32> {
+        self.handle(Request::RowInserts(requests)).await
+    }
+
+    /// Make a RowInsertRequests request to the database with hints.
+    pub async fn row_inserts_with_hints(
+        &self,
+        requests: RowInsertRequests,
+        hints: &[(&str, &str)],
+    ) -> Result<u32> {
+        let mut client = make_database_client(&self.client)?;
+        let request = self.to_rpc_request(Request::RowInserts(requests));
+
+        let mut request = tonic::Request::new(request);
+        let metadata = request.metadata_mut();
+        Self::put_hints(metadata, hints)?;
+
+        let response = client
+            .inner
+            .handle(request)
+            .await
+            .inspect_err(client.inspect_err("row_inserts_with_hints"))?
+            .into_inner();
         from_grpc_response(response)
     }
 
@@ -187,10 +237,16 @@ impl Database {
         Ok(())
     }
 
+    /// Make a request to the database.
     pub async fn handle(&self, request: Request) -> Result<u32> {
-        let mut client = make_database_client(&self.client)?.inner;
+        let mut client = make_database_client(&self.client)?;
         let request = self.to_rpc_request(request);
-        let response = client.handle(request).await?.into_inner();
+        let response = client
+            .inner
+            .handle(request)
+            .await
+            .inspect_err(client.inspect_err("handle"))?
+            .into_inner();
         from_grpc_response(response)
     }
 
@@ -202,7 +258,7 @@ impl Database {
         max_retries: u32,
         hints: &[(&str, &str)],
     ) -> Result<u32> {
-        let mut client = make_database_client(&self.client)?.inner;
+        let mut client = make_database_client(&self.client)?;
         let mut retries = 0;
 
         let request = self.to_rpc_request(request);
@@ -211,7 +267,11 @@ impl Database {
             let mut tonic_request = tonic::Request::new(request.clone());
             let metadata = tonic_request.metadata_mut();
             Self::put_hints(metadata, hints)?;
-            let raw_response = client.handle(tonic_request).await;
+            let raw_response = client
+                .inner
+                .handle(tonic_request)
+                .await
+                .inspect_err(client.inspect_err("handle"));
             match (raw_response, retries < max_retries) {
                 (Ok(resp), _) => return from_grpc_response(resp.into_inner()),
                 (Err(err), true) => {
@@ -221,12 +281,18 @@ impl Database {
                         retries += 1;
                         warn!("Retrying {} times with error = {:?}", retries, err);
                         continue;
+                    } else {
+                        error!(
+                            err; "Failed to send request to grpc handle, retries = {}, not retryable error, aborting",
+                            retries
+                        );
+                        return Err(err.into());
                     }
                 }
                 (Err(err), false) => {
                     error!(
-                        "Failed to send request to grpc handle after {} retries, error = {:?}",
-                        retries, err
+                        err; "Failed to send request to grpc handle after {} retries",
+                        retries,
                     );
                     return Err(err.into());
                 }
@@ -250,6 +316,7 @@ impl Database {
         }
     }
 
+    /// Executes a SQL query without any hints.
     pub async fn sql<S>(&self, sql: S) -> Result<Output>
     where
         S: AsRef<str>,
@@ -257,6 +324,7 @@ impl Database {
         self.sql_with_hint(sql, &[]).await
     }
 
+    /// Executes a SQL query with optional hints for query optimization.
     pub async fn sql_with_hint<S>(&self, sql: S, hints: &[(&str, &str)]) -> Result<Output>
     where
         S: AsRef<str>,
@@ -267,6 +335,7 @@ impl Database {
         self.do_get(request, hints).await
     }
 
+    /// Executes a logical plan directly without SQL parsing.
     pub async fn logical_plan(&self, logical_plan: Vec<u8>) -> Result<Output> {
         let request = Request::Query(QueryRequest {
             query: Some(Query::LogicalPlan(logical_plan)),
@@ -274,6 +343,7 @@ impl Database {
         self.do_get(request, &[]).await
     }
 
+    /// Creates a new table using the provided table expression.
     pub async fn create(&self, expr: CreateTableExpr) -> Result<Output> {
         let request = Request::Ddl(DdlRequest {
             expr: Some(DdlExpr::CreateTable(expr)),
@@ -281,6 +351,7 @@ impl Database {
         self.do_get(request, &[]).await
     }
 
+    /// Alters an existing table using the provided alter expression.
     pub async fn alter(&self, expr: AlterTableExpr) -> Result<Output> {
         let request = Request::Ddl(DdlRequest {
             expr: Some(DdlExpr::AlterTable(expr)),
@@ -321,7 +392,10 @@ impl Database {
         let mut flight_message_stream = flight_data_stream.map(move |flight_data| {
             flight_data
                 .map_err(Error::from)
-                .and_then(|data| decoder.try_decode(&data).context(ConvertFlightDataSnafu))
+                .and_then(|data| decoder.try_decode(&data).context(ConvertFlightDataSnafu))?
+                .context(IllegalFlightMessagesSnafu {
+                    reason: "none message",
+                })
         });
 
         let Some(first_flight_message) = flight_message_stream.next().await else {

src/client/src/error.rs

@@ -133,6 +133,13 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display("External error"))]
+    External {
+        #[snafu(implicit)]
+        location: Location,
+        source: BoxedError,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -154,6 +161,7 @@ impl ErrorExt for Error {
             Error::IllegalGrpcClientState { .. } => StatusCode::Unexpected,
             Error::InvalidTonicMetadataValue { .. } => StatusCode::InvalidArguments,
             Error::ConvertSchema { source, .. } => source.status_code(),
+            Error::External { source, .. } => source.status_code(),
         }
     }
 

src/client/src/inserter.rs

@@ -0,0 +1,60 @@
+// 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::time::Duration;
+
+use api::v1::RowInsertRequests;
+use humantime::format_duration;
+use store_api::mito_engine_options::{APPEND_MODE_KEY, TTL_KEY};
+
+use crate::error::Result;
+
+/// Context holds the catalog and schema information.
+pub struct Context<'a> {
+    /// The catalog name.
+    pub catalog: &'a str,
+    /// The schema name.
+    pub schema: &'a str,
+}
+
+/// Options for insert operations.
+#[derive(Debug, Clone, Copy)]
+pub struct InsertOptions {
+    /// Time-to-live for the inserted data.
+    pub ttl: Duration,
+    /// Whether to use append mode for the insert.
+    pub append_mode: bool,
+}
+
+impl InsertOptions {
+    /// Converts the insert options to a list of key-value string hints.
+    pub fn to_hints(&self) -> Vec<(&'static str, String)> {
+        vec![
+            (TTL_KEY, format_duration(self.ttl).to_string()),
+            (APPEND_MODE_KEY, self.append_mode.to_string()),
+        ]
+    }
+}
+
+/// [`Inserter`] allows different components to share a unified mechanism for inserting data.
+///
+/// An implementation may perform the insert locally (e.g., via a direct procedure call) or
+/// delegate/forward it to another node for processing (e.g., MetaSrv forwarding to an
+/// available Frontend).
+#[async_trait::async_trait]
+pub trait Inserter: Send + Sync {
+    async fn insert_rows(&self, context: &Context<'_>, requests: RowInsertRequests) -> Result<()>;
+
+    fn set_options(&mut self, options: &InsertOptions);
+}

src/client/src/lib.rs

@@ -19,6 +19,7 @@ pub mod client_manager;
 pub mod database;
 pub mod error;
 pub mod flow;
+pub mod inserter;
 pub mod load_balance;
 mod metrics;
 pub mod region;

src/client/src/region.rs

@@ -128,7 +128,10 @@ impl RegionRequester {
         let mut flight_message_stream = flight_data_stream.map(move |flight_data| {
             flight_data
                 .map_err(Error::from)
-                .and_then(|data| decoder.try_decode(&data).context(ConvertFlightDataSnafu))
+                .and_then(|data| decoder.try_decode(&data).context(ConvertFlightDataSnafu))?
+                .context(IllegalFlightMessagesSnafu {
+                    reason: "none message",
+                })
         });
 
         let Some(first_flight_message) = flight_message_stream.next().await else {
@@ -157,19 +160,70 @@ impl RegionRequester {
             let _span = tracing_context.attach(common_telemetry::tracing::info_span!(
                 "poll_flight_data_stream"
             ));
-            while let Some(flight_message) = flight_message_stream.next().await {
-                let flight_message = flight_message
-                    .map_err(BoxedError::new)
-                    .context(ExternalSnafu)?;
+
+            let mut buffered_message: Option<FlightMessage> = None;
+            let mut stream_ended = false;
+
+            while !stream_ended {
+                // get the next message from the buffered message or read from the flight message stream
+                let flight_message_item = if let Some(msg) = buffered_message.take() {
+                    Some(Ok(msg))
+                } else {
+                    flight_message_stream.next().await
+                };
+
+                let flight_message = match flight_message_item {
+                    Some(Ok(message)) => message,
+                    Some(Err(e)) => {
+                        yield Err(BoxedError::new(e)).context(ExternalSnafu);
+                        break;
+                    }
+                    None => break,
+                };
 
                 match flight_message {
                     FlightMessage::RecordBatch(record_batch) => {
-                        yield RecordBatch::try_from_df_record_batch(
+                        let result_to_yield = RecordBatch::try_from_df_record_batch(
                             schema_cloned.clone(),
                             record_batch,
-                        )
+                        );
+
+                        // get the next message from the stream. normally it should be a metrics message.
+                        if let Some(next_flight_message_result) = flight_message_stream.next().await
+                        {
+                            match next_flight_message_result {
+                                Ok(FlightMessage::Metrics(s)) => {
+                                    let m = serde_json::from_str(&s).ok().map(Arc::new);
+                                    metrics_ref.swap(m);
+                                }
+                                Ok(FlightMessage::RecordBatch(rb)) => {
+                                    // for some reason it's not a metrics message, so we need to buffer this record batch
+                                    // and yield it in the next iteration.
+                                    buffered_message = Some(FlightMessage::RecordBatch(rb));
+                                }
+                                Ok(_) => {
+                                    yield IllegalFlightMessagesSnafu {
+                                        reason: "A RecordBatch message can only be succeeded by a Metrics message or another RecordBatch message"
+                                    }
+                                    .fail()
+                                    .map_err(BoxedError::new)
+                                    .context(ExternalSnafu);
+                                    break;
+                                }
+                                Err(e) => {
+                                    yield Err(BoxedError::new(e)).context(ExternalSnafu);
+                                    break;
+                                }
+                            }
+                        } else {
+                            // the stream has ended
+                            stream_ended = true;
+                        }
+
+                        yield result_to_yield;
                     }
                     FlightMessage::Metrics(s) => {
+                        // just a branch in case of some metrics message comes after other things.
                         let m = serde_json::from_str(&s).ok().map(Arc::new);
                         metrics_ref.swap(m);
                         break;

src/cmd/Cargo.toml

@@ -38,6 +38,7 @@ common-config.workspace = true
 common-error.workspace = true
 common-grpc.workspace = true
 common-macro.workspace = true
+common-mem-prof.workspace = true
 common-meta.workspace = true
 common-options.workspace = true
 common-procedure.workspace = true

src/cmd/src/datanode/builder.rs

@@ -28,7 +28,7 @@ use tracing_appender::non_blocking::WorkerGuard;
 
 use crate::datanode::{DatanodeOptions, Instance, APP_NAME};
 use crate::error::{MetaClientInitSnafu, MissingConfigSnafu, Result, StartDatanodeSnafu};
-use crate::{create_resource_limit_metrics, log_versions};
+use crate::{create_resource_limit_metrics, log_versions, maybe_activate_heap_profile};
 
 /// Builder for Datanode instance.
 pub struct InstanceBuilder {
@@ -68,6 +68,7 @@ impl InstanceBuilder {
         );
 
         log_versions(verbose_version(), short_version(), APP_NAME);
+        maybe_activate_heap_profile(&dn_opts.memory);
         create_resource_limit_metrics(APP_NAME);
 
         plugins::setup_datanode_plugins(plugins, &opts.plugins, dn_opts)

src/cmd/src/flownode.rs

@@ -46,7 +46,7 @@ use crate::error::{
     MissingConfigSnafu, Result, ShutdownFlownodeSnafu, StartFlownodeSnafu,
 };
 use crate::options::{GlobalOptions, GreptimeOptions};
-use crate::{create_resource_limit_metrics, log_versions, App};
+use crate::{create_resource_limit_metrics, log_versions, maybe_activate_heap_profile, App};
 
 pub const APP_NAME: &str = "greptime-flownode";
 
@@ -280,6 +280,7 @@ impl StartCommand {
         );
 
         log_versions(verbose_version(), short_version(), APP_NAME);
+        maybe_activate_heap_profile(&opts.component.memory);
         create_resource_limit_metrics(APP_NAME);
 
         info!("Flownode start command: {:#?}", self);
@@ -374,7 +375,9 @@ impl StartCommand {
             meta_client.clone(),
             flow_auth_header,
             opts.query.clone(),
-        );
+            opts.flow.batching_mode.clone(),
+        )
+        .context(StartFlownodeSnafu)?;
         let frontend_client = Arc::new(frontend_client);
         let flownode_builder = FlownodeBuilder::new(
             opts.clone(),

src/cmd/src/frontend.rs

@@ -47,7 +47,7 @@ use tracing_appender::non_blocking::WorkerGuard;
 
 use crate::error::{self, Result};
 use crate::options::{GlobalOptions, GreptimeOptions};
-use crate::{create_resource_limit_metrics, log_versions, App};
+use crate::{create_resource_limit_metrics, log_versions, maybe_activate_heap_profile, App};
 
 type FrontendOptions = GreptimeOptions<frontend::frontend::FrontendOptions>;
 
@@ -279,10 +279,11 @@ impl StartCommand {
             &opts.component.logging,
             &opts.component.tracing,
             opts.component.node_id.clone(),
-            opts.component.slow_query.as_ref(),
+            Some(&opts.component.slow_query),
         );
 
         log_versions(verbose_version(), short_version(), APP_NAME);
+        maybe_activate_heap_profile(&opts.component.memory);
         create_resource_limit_metrics(APP_NAME);
 
         info!("Frontend start command: {:#?}", self);

src/cmd/src/lib.rs

@@ -15,7 +15,10 @@
 #![feature(assert_matches, let_chains)]
 
 use async_trait::async_trait;
-use common_telemetry::{error, info};
+use common_error::ext::ErrorExt;
+use common_error::status_code::StatusCode;
+use common_mem_prof::activate_heap_profile;
+use common_telemetry::{error, info, warn};
 use stat::{get_cpu_limit, get_memory_limit};
 
 use crate::error::Result;
@@ -145,3 +148,20 @@ fn log_env_flags() {
         info!("argument: {}", argument);
     }
 }
+
+pub fn maybe_activate_heap_profile(memory_options: &common_options::memory::MemoryOptions) {
+    if memory_options.enable_heap_profiling {
+        match activate_heap_profile() {
+            Ok(()) => {
+                info!("Heap profile is active");
+            }
+            Err(err) => {
+                if err.status_code() == StatusCode::Unsupported {
+                    info!("Heap profile is not supported");
+                } else {
+                    warn!(err; "Failed to activate heap profile");
+                }
+            }
+        }
+    }
+}

src/cmd/src/metasrv.rs

@@ -30,7 +30,7 @@ use tracing_appender::non_blocking::WorkerGuard;
 
 use crate::error::{self, LoadLayeredConfigSnafu, Result, StartMetaServerSnafu};
 use crate::options::{GlobalOptions, GreptimeOptions};
-use crate::{create_resource_limit_metrics, log_versions, App};
+use crate::{create_resource_limit_metrics, log_versions, maybe_activate_heap_profile, App};
 
 type MetasrvOptions = GreptimeOptions<meta_srv::metasrv::MetasrvOptions>;
 
@@ -325,6 +325,7 @@ impl StartCommand {
         );
 
         log_versions(verbose_version(), short_version(), APP_NAME);
+        maybe_activate_heap_profile(&opts.component.memory);
         create_resource_limit_metrics(APP_NAME);
 
         info!("Metasrv start command: {:#?}", self);

src/cmd/src/standalone.rs

@@ -34,17 +34,19 @@ use common_meta::cluster::{NodeInfo, NodeStatus};
 use common_meta::datanode::RegionStat;
 use common_meta::ddl::flow_meta::FlowMetadataAllocator;
 use common_meta::ddl::table_meta::TableMetadataAllocator;
-use common_meta::ddl::{DdlContext, NoopRegionFailureDetectorControl, ProcedureExecutorRef};
+use common_meta::ddl::{DdlContext, NoopRegionFailureDetectorControl};
 use common_meta::ddl_manager::DdlManager;
 use common_meta::key::flow::flow_state::FlowStat;
 use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
 use common_meta::peer::Peer;
+use common_meta::procedure_executor::LocalProcedureExecutor;
 use common_meta::region_keeper::MemoryRegionKeeper;
 use common_meta::region_registry::LeaderRegionRegistry;
 use common_meta::sequence::SequenceBuilder;
 use common_meta::wal_options_allocator::{build_wal_options_allocator, WalOptionsAllocatorRef};
+use common_options::memory::MemoryOptions;
 use common_procedure::{ProcedureInfo, ProcedureManagerRef};
 use common_telemetry::info;
 use common_telemetry::logging::{
@@ -83,7 +85,7 @@ use tracing_appender::non_blocking::WorkerGuard;
 
 use crate::error::{Result, StartFlownodeSnafu};
 use crate::options::{GlobalOptions, GreptimeOptions};
-use crate::{create_resource_limit_metrics, error, log_versions, App};
+use crate::{create_resource_limit_metrics, error, log_versions, maybe_activate_heap_profile, App};
 
 pub const APP_NAME: &str = "greptime-standalone";
 
@@ -155,8 +157,9 @@ pub struct StandaloneOptions {
     pub init_regions_in_background: bool,
     pub init_regions_parallelism: usize,
     pub max_in_flight_write_bytes: Option<ReadableSize>,
-    pub slow_query: Option<SlowQueryOptions>,
+    pub slow_query: SlowQueryOptions,
     pub query: QueryOptions,
+    pub memory: MemoryOptions,
 }
 
 impl Default for StandaloneOptions {
@@ -188,8 +191,9 @@ impl Default for StandaloneOptions {
             init_regions_in_background: false,
             init_regions_parallelism: 16,
             max_in_flight_write_bytes: None,
-            slow_query: Some(SlowQueryOptions::default()),
+            slow_query: SlowQueryOptions::default(),
             query: QueryOptions::default(),
+            memory: MemoryOptions::default(),
         }
     }
 }
@@ -482,10 +486,11 @@ impl StartCommand {
             &opts.component.logging,
             &opts.component.tracing,
             None,
-            opts.component.slow_query.as_ref(),
+            Some(&opts.component.slow_query),
         );
 
         log_versions(verbose_version(), short_version(), APP_NAME);
+        maybe_activate_heap_profile(&opts.component.memory);
         create_resource_limit_metrics(APP_NAME);
 
         info!("Standalone start command: {:#?}", self);
@@ -636,9 +641,8 @@ impl StartCommand {
             flow_metadata_allocator: flow_metadata_allocator.clone(),
             region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl),
         };
-        let procedure_manager_c = procedure_manager.clone();
 
-        let ddl_manager = DdlManager::try_new(ddl_context, procedure_manager_c, true)
+        let ddl_manager = DdlManager::try_new(ddl_context, procedure_manager.clone(), true)
             .context(error::InitDdlManagerSnafu)?;
         #[cfg(feature = "enterprise")]
         let ddl_manager = {
@@ -646,7 +650,11 @@ impl StartCommand {
                 plugins.get();
             ddl_manager.with_trigger_ddl_manager(trigger_ddl_manager)
         };
-        let ddl_task_executor: ProcedureExecutorRef = Arc::new(ddl_manager);
+
+        let procedure_executor = Arc::new(LocalProcedureExecutor::new(
+            Arc::new(ddl_manager),
+            procedure_manager.clone(),
+        ));
 
         let fe_instance = FrontendBuilder::new(
             fe_opts.clone(),
@@ -654,7 +662,7 @@ impl StartCommand {
             layered_cache_registry.clone(),
             catalog_manager.clone(),
             node_manager.clone(),
-            ddl_task_executor.clone(),
+            procedure_executor.clone(),
             process_manager,
         )
         .with_plugin(plugins.clone())
@@ -679,7 +687,7 @@ impl StartCommand {
             catalog_manager.clone(),
             kv_backend.clone(),
             layered_cache_registry.clone(),
-            ddl_task_executor.clone(),
+            procedure_executor,
             node_manager,
         )
         .await
@@ -821,10 +829,12 @@ impl InformationExtension for StandaloneInformationExtension {
                     memtable_size: region_stat.memtable_size,
                     manifest_size: region_stat.manifest_size,
                     sst_size: region_stat.sst_size,
+                    sst_num: region_stat.sst_num,
                     index_size: region_stat.index_size,
                     region_manifest: region_stat.manifest.into(),
                     data_topic_latest_entry_id: region_stat.data_topic_latest_entry_id,
                     metadata_topic_latest_entry_id: region_stat.metadata_topic_latest_entry_id,
+                    write_bytes: 0,
                 }
             })
             .collect::<Vec<_>>();

src/cmd/tests/load_config_test.rs

@@ -34,6 +34,7 @@ use query::options::QueryOptions;
 use servers::export_metrics::ExportMetricsOption;
 use servers::grpc::GrpcOptions;
 use servers::http::HttpOptions;
+use servers::tls::{TlsMode, TlsOption};
 use store_api::path_utils::WAL_DIR;
 
 #[allow(deprecated)]
@@ -190,6 +191,13 @@ fn test_load_metasrv_example_config() {
                 remote_write: Some(Default::default()),
                 ..Default::default()
             },
+            backend_tls: Some(TlsOption {
+                mode: TlsMode::Prefer,
+                cert_path: String::new(),
+                key_path: String::new(),
+                ca_cert_path: String::new(),
+                watch: false,
+            }),
             ..Default::default()
         },
         ..Default::default()
@@ -225,7 +233,10 @@ fn test_load_flownode_example_config() {
             heartbeat: Default::default(),
             // flownode deliberately use a slower query parallelism
             // to avoid overwhelming the frontend with too many queries
-            query: QueryOptions { parallelism: 1 },
+            query: QueryOptions {
+                parallelism: 1,
+                allow_query_fallback: false,
+            },
             meta_client: Some(MetaClientOptions {
                 metasrv_addrs: vec!["127.0.0.1:3002".to_string()],
                 timeout: Duration::from_secs(3),
@@ -242,6 +253,7 @@ fn test_load_flownode_example_config() {
                 ..Default::default()
             },
             user_provider: None,
+            memory: Default::default(),
         },
         ..Default::default()
     };
@@ -295,6 +307,7 @@ fn test_load_standalone_example_config() {
                 cors_allowed_origins: vec!["https://example.com".to_string()],
                 ..Default::default()
             },
+
             ..Default::default()
         },
         ..Default::default()

src/common/base/src/lib.rs

@@ -20,6 +20,7 @@ pub mod range_read;
 #[allow(clippy::all)]
 pub mod readable_size;
 pub mod secrets;
+pub mod serde;
 
 pub type AffectedRows = usize;
 

src/common/base/src/serde.rs

@@ -12,16 +12,20 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! Format to store in parquet.
-//!
-//! We store two additional internal columns at last:
-//! - `__sequence`, the sequence number of a row. Type: uint64
-//! - `__op_type`, the op type of the row. Type: uint8
-//!
-//! We store other columns in the same order as [RegionMetadata::field_columns()](store_api::metadata::RegionMetadata::field_columns()).
-//!
+use serde::{Deserialize, Deserializer};
 
-/// Number of columns that have fixed positions.
-///
-/// Contains all internal columns.
-pub(crate) const PLAIN_FIXED_POS_COLUMN_NUM: usize = 2;
+/// Deserialize an empty string as the default value.
+pub fn empty_string_as_default<'de, D, T>(deserializer: D) -> Result<T, D::Error>
+where
+    D: Deserializer<'de>,
+    T: Default + Deserialize<'de>,
+{
+    let s = String::deserialize(deserializer)?;
+
+    if s.is_empty() {
+        Ok(T::default())
+    } else {
+        T::deserialize(serde::de::value::StringDeserializer::<D::Error>::new(s))
+            .map_err(serde::de::Error::custom)
+    }
+}

src/common/catalog/src/consts.rs

@@ -104,8 +104,6 @@ pub const INFORMATION_SCHEMA_PROCEDURE_INFO_TABLE_ID: u32 = 34;
 pub const INFORMATION_SCHEMA_REGION_STATISTICS_TABLE_ID: u32 = 35;
 /// id for information_schema.process_list
 pub const INFORMATION_SCHEMA_PROCESS_LIST_TABLE_ID: u32 = 36;
-/// id for information_schema.trigger_list (for greptimedb trigger)
-pub const INFORMATION_SCHEMA_TRIGGER_TABLE_ID: u32 = 37;
 
 // ----- End of information_schema tables -----
 

src/common/datasource/Cargo.toml

@@ -25,19 +25,17 @@ common-error.workspace = true
 common-macro.workspace = true
 common-recordbatch.workspace = true
 common-runtime.workspace = true
+common-telemetry.workspace = true
 datafusion.workspace = true
+datafusion-orc.workspace = true
 datatypes.workspace = true
-derive_builder.workspace = true
 futures.workspace = true
 lazy_static.workspace = true
 object-store.workspace = true
 object_store_opendal.workspace = true
-orc-rust = { git = "https://github.com/datafusion-contrib/orc-rust", rev = "3134cab581a8e91b942d6a23aca2916ea965f6bb", default-features = false, features = [
-    "async",
-] }
+orc-rust = { version = "0.6.3", default-features = false, features = ["async"] }
 parquet.workspace = true
 paste.workspace = true
-rand.workspace = true
 regex = "1.7"
 serde.workspace = true
 snafu.workspace = true
@@ -47,6 +45,4 @@ tokio-util.workspace = true
 url = "2.3"
 
 [dev-dependencies]
-common-telemetry.workspace = true
 common-test-util.workspace = true
-uuid.workspace = true

src/common/datasource/src/file_format/orc.rs

@@ -12,16 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::Arc;
-
-use arrow_schema::{ArrowError, Schema, SchemaRef};
+use arrow_schema::Schema;
 use async_trait::async_trait;
 use bytes::Bytes;
-use common_recordbatch::adapter::RecordBatchStreamTypeAdapter;
-use datafusion::datasource::physical_plan::{FileMeta, FileOpenFuture, FileOpener};
-use datafusion::error::{DataFusionError, Result as DfResult};
 use futures::future::BoxFuture;
-use futures::{FutureExt, StreamExt, TryStreamExt};
+use futures::FutureExt;
 use object_store::ObjectStore;
 use orc_rust::arrow_reader::ArrowReaderBuilder;
 use orc_rust::async_arrow_reader::ArrowStreamReader;
@@ -97,67 +92,6 @@ impl FileFormat for OrcFormat {
     }
 }
 
-#[derive(Debug, Clone)]
-pub struct OrcOpener {
-    object_store: Arc<ObjectStore>,
-    output_schema: SchemaRef,
-    projection: Option<Vec<usize>>,
-}
-
-impl OrcOpener {
-    pub fn new(
-        object_store: ObjectStore,
-        output_schema: SchemaRef,
-        projection: Option<Vec<usize>>,
-    ) -> Self {
-        Self {
-            object_store: Arc::from(object_store),
-            output_schema,
-            projection,
-        }
-    }
-}
-
-impl FileOpener for OrcOpener {
-    fn open(&self, meta: FileMeta) -> DfResult<FileOpenFuture> {
-        let object_store = self.object_store.clone();
-        let projected_schema = if let Some(projection) = &self.projection {
-            let projected_schema = self
-                .output_schema
-                .project(projection)
-                .map_err(|e| DataFusionError::External(Box::new(e)))?;
-            Arc::new(projected_schema)
-        } else {
-            self.output_schema.clone()
-        };
-        let projection = self.projection.clone();
-        Ok(Box::pin(async move {
-            let path = meta.location().to_string();
-
-            let meta = object_store
-                .stat(&path)
-                .await
-                .map_err(|e| DataFusionError::External(Box::new(e)))?;
-
-            let reader = object_store
-                .reader(&path)
-                .await
-                .map_err(|e| DataFusionError::External(Box::new(e)))?;
-
-            let stream_reader =
-                new_orc_stream_reader(ReaderAdapter::new(reader, meta.content_length()))
-                    .await
-                    .map_err(|e| DataFusionError::External(Box::new(e)))?;
-
-            let stream =
-                RecordBatchStreamTypeAdapter::new(projected_schema, stream_reader, projection);
-
-            let adopted = stream.map_err(|e| ArrowError::ExternalError(Box::new(e)));
-            Ok(adopted.boxed())
-        }))
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use common_test_util::find_workspace_path;

src/common/datasource/src/file_format/parquet.rs

@@ -31,6 +31,7 @@ use datatypes::schema::SchemaRef;
 use futures::future::BoxFuture;
 use futures::StreamExt;
 use object_store::{FuturesAsyncReader, ObjectStore};
+use parquet::arrow::arrow_reader::ArrowReaderOptions;
 use parquet::arrow::AsyncArrowWriter;
 use parquet::basic::{Compression, Encoding, ZstdLevel};
 use parquet::file::properties::{WriterProperties, WriterPropertiesBuilder};
@@ -65,7 +66,7 @@ impl FileFormat for ParquetFormat {
             .compat();
 
         let metadata = reader
-            .get_metadata()
+            .get_metadata(None)
             .await
             .context(error::ReadParquetSnafuSnafu)?;
 
@@ -146,7 +147,7 @@ impl LazyParquetFileReader {
 impl AsyncFileReader for LazyParquetFileReader {
     fn get_bytes(
         &mut self,
-        range: std::ops::Range<usize>,
+        range: std::ops::Range<u64>,
     ) -> BoxFuture<'_, ParquetResult<bytes::Bytes>> {
         Box::pin(async move {
             self.maybe_initialize()
@@ -157,13 +158,16 @@ impl AsyncFileReader for LazyParquetFileReader {
         })
     }
 
-    fn get_metadata(&mut self) -> BoxFuture<'_, ParquetResult<Arc<ParquetMetaData>>> {
+    fn get_metadata<'a>(
+        &'a mut self,
+        options: Option<&'a ArrowReaderOptions>,
+    ) -> BoxFuture<'a, parquet::errors::Result<Arc<ParquetMetaData>>> {
         Box::pin(async move {
             self.maybe_initialize()
                 .await
                 .map_err(|e| ParquetError::External(Box::new(e)))?;
             // Safety: Must initialized
-            self.reader.as_mut().unwrap().get_metadata().await
+            self.reader.as_mut().unwrap().get_metadata(options).await
         })
     }
 }

src/common/datasource/src/file_format/tests.rs

@@ -19,35 +19,39 @@ use std::vec;
 
 use common_test_util::find_workspace_path;
 use datafusion::assert_batches_eq;
-use datafusion::datasource::file_format::file_compression_type::FileCompressionType;
 use datafusion::datasource::physical_plan::{
-    CsvConfig, CsvOpener, FileOpener, FileScanConfig, FileStream, JsonOpener, ParquetExec,
+    CsvSource, FileScanConfig, FileSource, FileStream, JsonSource, ParquetSource,
 };
+use datafusion::datasource::source::DataSourceExec;
 use datafusion::execution::context::TaskContext;
 use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion::prelude::SessionContext;
+use datafusion_orc::OrcSource;
 use futures::StreamExt;
+use object_store::ObjectStore;
 
 use super::FORMAT_TYPE;
-use crate::file_format::orc::{OrcFormat, OrcOpener};
 use crate::file_format::parquet::DefaultParquetFileReaderFactory;
-use crate::file_format::{FileFormat, Format};
+use crate::file_format::{FileFormat, Format, OrcFormat};
 use crate::test_util::{scan_config, test_basic_schema, test_store};
 use crate::{error, test_util};
 
-struct Test<'a, T: FileOpener> {
+struct Test<'a> {
     config: FileScanConfig,
-    opener: T,
+    file_source: Arc<dyn FileSource>,
     expected: Vec<&'a str>,
 }
 
-impl<T: FileOpener> Test<'_, T> {
-    pub async fn run(self) {
+impl Test<'_> {
+    async fn run(self, store: &ObjectStore) {
+        let store = Arc::new(object_store_opendal::OpendalStore::new(store.clone()));
+        let file_opener = self.file_source.create_file_opener(store, &self.config, 0);
+
         let result = FileStream::new(
             &self.config,
             0,
-            self.opener,
+            file_opener,
             &ExecutionPlanMetricsSet::new(),
         )
         .unwrap()
@@ -62,26 +66,16 @@ impl<T: FileOpener> Test<'_, T> {
 #[tokio::test]
 async fn test_json_opener() {
     let store = test_store("/");
-    let store = Arc::new(object_store_opendal::OpendalStore::new(store));
-
     let schema = test_basic_schema();
-
-    let json_opener = || {
-        JsonOpener::new(
-            test_util::TEST_BATCH_SIZE,
-            schema.clone(),
-            FileCompressionType::UNCOMPRESSED,
-            store.clone(),
-        )
-    };
+    let file_source = Arc::new(JsonSource::new()).with_batch_size(test_util::TEST_BATCH_SIZE);
 
     let path = &find_workspace_path("/src/common/datasource/tests/json/basic.json")
         .display()
         .to_string();
     let tests = [
         Test {
-            config: scan_config(schema.clone(), None, path),
-            opener: json_opener(),
+            config: scan_config(schema.clone(), None, path, file_source.clone()),
+            file_source: file_source.clone(),
             expected: vec![
                 "+-----+-------+",
                 "| num | str   |",
@@ -93,8 +87,8 @@ async fn test_json_opener() {
             ],
         },
         Test {
-            config: scan_config(schema.clone(), Some(1), path),
-            opener: json_opener(),
+            config: scan_config(schema, Some(1), path, file_source.clone()),
+            file_source,
             expected: vec![
                 "+-----+------+",
                 "| num | str  |",
@@ -106,37 +100,26 @@ async fn test_json_opener() {
     ];
 
     for test in tests {
-        test.run().await;
+        test.run(&store).await;
     }
 }
 
 #[tokio::test]
 async fn test_csv_opener() {
     let store = test_store("/");
-    let store = Arc::new(object_store_opendal::OpendalStore::new(store));
-
     let schema = test_basic_schema();
     let path = &find_workspace_path("/src/common/datasource/tests/csv/basic.csv")
         .display()
         .to_string();
-    let csv_config = Arc::new(CsvConfig::new(
-        test_util::TEST_BATCH_SIZE,
-        schema.clone(),
-        None,
-        true,
-        b',',
-        b'"',
-        None,
-        store,
-        None,
-    ));
 
-    let csv_opener = || CsvOpener::new(csv_config.clone(), FileCompressionType::UNCOMPRESSED);
+    let file_source = CsvSource::new(true, b',', b'"')
+        .with_batch_size(test_util::TEST_BATCH_SIZE)
+        .with_schema(schema.clone());
 
     let tests = [
         Test {
-            config: scan_config(schema.clone(), None, path),
-            opener: csv_opener(),
+            config: scan_config(schema.clone(), None, path, file_source.clone()),
+            file_source: file_source.clone(),
             expected: vec![
                 "+-----+-------+",
                 "| num | str   |",
@@ -148,8 +131,8 @@ async fn test_csv_opener() {
             ],
         },
         Test {
-            config: scan_config(schema.clone(), Some(1), path),
-            opener: csv_opener(),
+            config: scan_config(schema, Some(1), path, file_source.clone()),
+            file_source,
             expected: vec![
                 "+-----+------+",
                 "| num | str  |",
@@ -161,7 +144,7 @@ async fn test_csv_opener() {
     ];
 
     for test in tests {
-        test.run().await;
+        test.run(&store).await;
     }
 }
 
@@ -174,12 +157,12 @@ async fn test_parquet_exec() {
     let path = &find_workspace_path("/src/common/datasource/tests/parquet/basic.parquet")
         .display()
         .to_string();
-    let base_config = scan_config(schema.clone(), None, path);
 
-    let exec = ParquetExec::builder(base_config)
-        .with_parquet_file_reader_factory(Arc::new(DefaultParquetFileReaderFactory::new(store)))
-        .build();
+    let parquet_source = ParquetSource::default()
+        .with_parquet_file_reader_factory(Arc::new(DefaultParquetFileReaderFactory::new(store)));
 
+    let config = scan_config(schema, None, path, Arc::new(parquet_source));
+    let exec = DataSourceExec::from_data_source(config);
     let ctx = SessionContext::new();
 
     let context = Arc::new(TaskContext::from(&ctx));
@@ -208,20 +191,18 @@ async fn test_parquet_exec() {
 
 #[tokio::test]
 async fn test_orc_opener() {
-    let root = find_workspace_path("/src/common/datasource/tests/orc")
+    let path = &find_workspace_path("/src/common/datasource/tests/orc/test.orc")
         .display()
         .to_string();
-    let store = test_store(&root);
-    let schema = OrcFormat.infer_schema(&store, "test.orc").await.unwrap();
-    let schema = Arc::new(schema);
 
-    let orc_opener = OrcOpener::new(store.clone(), schema.clone(), None);
-    let path = "test.orc";
+    let store = test_store("/");
+    let schema = Arc::new(OrcFormat.infer_schema(&store, path).await.unwrap());
+    let file_source = Arc::new(OrcSource::default());
 
     let tests = [
         Test {
-            config: scan_config(schema.clone(), None, path),
-            opener: orc_opener.clone(),
+            config: scan_config(schema.clone(), None, path, file_source.clone()),
+            file_source: file_source.clone(),
             expected: vec![
             "+----------+-----+-------+------------+-----+-----+-------+--------------------+------------------------+-----------+---------------+------------+----------------+---------------+-------------------+--------------+---------------+---------------+----------------------------+-------------+",
             "| double_a | a   | b     | str_direct | d   | e   | f     | int_short_repeated | int_neg_short_repeated | int_delta | int_neg_delta | int_direct | int_neg_direct | bigint_direct | bigint_neg_direct | bigint_other | utf8_increase | utf8_decrease | timestamp_simple           | date_simple |",
@@ -235,8 +216,8 @@ async fn test_orc_opener() {
             ],
         },
         Test {
-            config: scan_config(schema.clone(), Some(1), path),
-            opener: orc_opener.clone(),
+            config: scan_config(schema.clone(), Some(1), path, file_source.clone()),
+            file_source,
             expected: vec![
                 "+----------+-----+------+------------+---+-----+-------+--------------------+------------------------+-----------+---------------+------------+----------------+---------------+-------------------+--------------+---------------+---------------+-------------------------+-------------+",
                 "| double_a | a   | b    | str_direct | d | e   | f     | int_short_repeated | int_neg_short_repeated | int_delta | int_neg_delta | int_direct | int_neg_direct | bigint_direct | bigint_neg_direct | bigint_other | utf8_increase | utf8_decrease | timestamp_simple        | date_simple |",
@@ -248,7 +229,7 @@ async fn test_orc_opener() {
     ];
 
     for test in tests {
-        test.run().await;
+        test.run(&store).await;
     }
 }
 

src/common/datasource/src/lib.rs

@@ -21,6 +21,7 @@ pub mod error;
 pub mod file_format;
 pub mod lister;
 pub mod object_store;
+pub mod parquet_writer;
 pub mod share_buffer;
 #[cfg(test)]
 pub mod test_util;

src/common/datasource/src/object_store/oss.rs

@@ -77,6 +77,11 @@ pub fn build_oss_backend(
 
     let op = ObjectStore::new(builder)
         .context(error::BuildBackendSnafu)?
+        .layer(
+            object_store::layers::RetryLayer::new()
+                .with_jitter()
+                .with_notify(object_store::util::PrintDetailedError),
+        )
         .layer(object_store::layers::LoggingLayer::default())
         .layer(object_store::layers::TracingLayer)
         .layer(object_store::layers::build_prometheus_metrics_layer(true))

src/common/datasource/src/object_store/s3.rs

@@ -85,6 +85,11 @@ pub fn build_s3_backend(
     // TODO(weny): Consider finding a better way to eliminate duplicate code.
     Ok(ObjectStore::new(builder)
         .context(error::BuildBackendSnafu)?
+        .layer(
+            object_store::layers::RetryLayer::new()
+                .with_jitter()
+                .with_notify(object_store::util::PrintDetailedError),
+        )
         .layer(object_store::layers::LoggingLayer::new(
             DefaultLoggingInterceptor,
         ))

src/common/datasource/src/parquet_writer.rs

@@ -0,0 +1,52 @@
+// 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 bytes::Bytes;
+use futures::future::BoxFuture;
+use object_store::Writer;
+use parquet::arrow::async_writer::AsyncFileWriter;
+use parquet::errors::ParquetError;
+
+/// Bridges opendal [Writer] with parquet [AsyncFileWriter].
+pub struct AsyncWriter {
+    inner: Writer,
+}
+
+impl AsyncWriter {
+    /// Create a [`AsyncWriter`] by given [`Writer`].
+    pub fn new(writer: Writer) -> Self {
+        Self { inner: writer }
+    }
+}
+
+impl AsyncFileWriter for AsyncWriter {
+    fn write(&mut self, bs: Bytes) -> BoxFuture<'_, parquet::errors::Result<()>> {
+        Box::pin(async move {
+            self.inner
+                .write(bs)
+                .await
+                .map_err(|err| ParquetError::External(Box::new(err)))
+        })
+    }
+
+    fn complete(&mut self) -> BoxFuture<'_, parquet::errors::Result<()>> {
+        Box::pin(async move {
+            self.inner
+                .close()
+                .await
+                .map(|_| ())
+                .map_err(|err| ParquetError::External(Box::new(err)))
+        })
+    }
+}

src/common/datasource/src/test_util.rs

@@ -16,12 +16,12 @@ use std::sync::Arc;
 
 use arrow_schema::{DataType, Field, Schema, SchemaRef};
 use common_test_util::temp_dir::{create_temp_dir, TempDir};
-use datafusion::common::{Constraints, Statistics};
 use datafusion::datasource::file_format::file_compression_type::FileCompressionType;
 use datafusion::datasource::listing::PartitionedFile;
 use datafusion::datasource::object_store::ObjectStoreUrl;
 use datafusion::datasource::physical_plan::{
-    CsvConfig, CsvOpener, FileScanConfig, FileStream, JsonOpener,
+    CsvSource, FileGroup, FileScanConfig, FileScanConfigBuilder, FileSource, FileStream,
+    JsonOpener, JsonSource,
 };
 use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
 use object_store::services::Fs;
@@ -68,21 +68,20 @@ pub fn test_basic_schema() -> SchemaRef {
     Arc::new(schema)
 }
 
-pub fn scan_config(file_schema: SchemaRef, limit: Option<usize>, filename: &str) -> FileScanConfig {
+pub(crate) fn scan_config(
+    file_schema: SchemaRef,
+    limit: Option<usize>,
+    filename: &str,
+    file_source: Arc<dyn FileSource>,
+) -> FileScanConfig {
     // object_store only recognize the Unix style path, so make it happy.
     let filename = &filename.replace('\\', "/");
-    let statistics = Statistics::new_unknown(file_schema.as_ref());
-    FileScanConfig {
-        object_store_url: ObjectStoreUrl::parse("empty://").unwrap(), // won't be used
-        file_schema,
-        file_groups: vec![vec![PartitionedFile::new(filename.to_string(), 10)]],
-        constraints: Constraints::empty(),
-        statistics,
-        projection: None,
-        limit,
-        table_partition_cols: vec![],
-        output_ordering: vec![],
-    }
+    let file_group = FileGroup::new(vec![PartitionedFile::new(filename.to_string(), 4096)]);
+
+    FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(), file_schema, file_source)
+        .with_file_group(file_group)
+        .with_limit(limit)
+        .build()
 }
 
 pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usize) -> usize) {
@@ -99,9 +98,14 @@ pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usi
 
     let size = store.read(origin_path).await.unwrap().len();
 
-    let config = scan_config(schema.clone(), None, origin_path);
-
-    let stream = FileStream::new(&config, 0, json_opener, &ExecutionPlanMetricsSet::new()).unwrap();
+    let config = scan_config(schema, None, origin_path, Arc::new(JsonSource::new()));
+    let stream = FileStream::new(
+        &config,
+        0,
+        Arc::new(json_opener),
+        &ExecutionPlanMetricsSet::new(),
+    )
+    .unwrap();
 
     let (tmp_store, dir) = test_tmp_store("test_stream_to_json");
 
@@ -127,24 +131,17 @@ pub async fn setup_stream_to_csv_test(origin_path: &str, threshold: impl Fn(usiz
 
     let schema = test_basic_schema();
 
-    let csv_config = Arc::new(CsvConfig::new(
-        TEST_BATCH_SIZE,
-        schema.clone(),
-        None,
-        true,
-        b',',
-        b'"',
-        None,
-        Arc::new(object_store_opendal::OpendalStore::new(store.clone())),
-        None,
-    ));
-
-    let csv_opener = CsvOpener::new(csv_config, FileCompressionType::UNCOMPRESSED);
-
+    let csv_source = CsvSource::new(true, b',', b'"')
+        .with_schema(schema.clone())
+        .with_batch_size(TEST_BATCH_SIZE);
+    let config = scan_config(schema, None, origin_path, csv_source.clone());
     let size = store.read(origin_path).await.unwrap().len();
 
-    let config = scan_config(schema.clone(), None, origin_path);
-
+    let csv_opener = csv_source.create_file_opener(
+        Arc::new(object_store_opendal::OpendalStore::new(store.clone())),
+        &config,
+        0,
+    );
     let stream = FileStream::new(&config, 0, csv_opener, &ExecutionPlanMetricsSet::new()).unwrap();
 
     let (tmp_store, dir) = test_tmp_store("test_stream_to_csv");

src/common/error/tests/ext.rs

@@ -84,24 +84,33 @@ fn test_to_string() {
     assert_eq!(result.unwrap_err().to_string(), "<root cause>");
 }
 
+fn normalize_path(s: &str) -> String {
+    s.replace('\\', "/")
+}
+
 #[test]
 fn test_debug_format() {
     let result = normal_error();
     let debug_output = format!("{:?}", result.unwrap_err());
-    let normalized_output = debug_output.replace('\\', "/");
+
     assert_eq!(
-        normalized_output,
-        r#"0: A normal error with "display" attribute, message "blabla", at src/common/error/tests/ext.rs:55:22
-1: PlainError { msg: "<root cause>", status_code: Unexpected }"#
+        normalize_path(&debug_output),
+        format!(
+            r#"0: A normal error with "display" attribute, message "blabla", at {}:55:22
+1: PlainError {{ msg: "<root cause>", status_code: Unexpected }}"#,
+            normalize_path(file!())
+        )
     );
 
     let result = transparent_error();
     let debug_output = format!("{:?}", result.unwrap_err());
-    let normalized_output = debug_output.replace('\\', "/");
     assert_eq!(
-        normalized_output,
-        r#"0: <transparent>, at src/common/error/tests/ext.rs:60:5
-1: PlainError { msg: "<root cause>", status_code: Unexpected }"#
+        normalize_path(&debug_output),
+        format!(
+            r#"0: <transparent>, at {}:60:5
+1: PlainError {{ msg: "<root cause>", status_code: Unexpected }}"#,
+            normalize_path(file!())
+        )
     );
 }
 

src/common/event-recorder/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "common-event-recorder"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[dependencies]
+api.workspace = true
+async-trait.workspace = true
+backon.workspace = true
+common-error.workspace = true
+common-macro.workspace = true
+common-telemetry.workspace = true
+common-time.workspace = true
+humantime.workspace = true
+humantime-serde.workspace = true
+itertools.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+snafu.workspace = true
+store-api.workspace = true
+tokio.workspace = true
+tokio-util.workspace = true
+
+[lints]
+workspace = true

src/common/event-recorder/src/error.rs

@@ -0,0 +1,73 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use api::v1::ColumnSchema;
+use common_error::ext::{BoxedError, ErrorExt};
+use common_error::status_code::StatusCode;
+use common_macro::stack_trace_debug;
+use snafu::{Location, Snafu};
+
+#[derive(Snafu)]
+#[snafu(visibility(pub))]
+#[stack_trace_debug]
+pub enum Error {
+    #[snafu(display("Mismatched schema, expected: {:?}, actual: {:?}", expected, actual))]
+    MismatchedSchema {
+        #[snafu(implicit)]
+        location: Location,
+        expected: Vec<ColumnSchema>,
+        actual: Vec<ColumnSchema>,
+    },
+
+    #[snafu(display("Failed to serialize event"))]
+    SerializeEvent {
+        #[snafu(source)]
+        error: serde_json::error::Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Failed to insert events"))]
+    InsertEvents {
+        // BoxedError is utilized here to prevent introducing a circular dependency that would arise from directly referencing `client::error::Error`.
+        source: BoxedError,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Keyvalue backend error"))]
+    KvBackend {
+        // BoxedError is utilized here to prevent introducing a circular dependency that would arise from directly referencing `common_meta::error::Error`.
+        source: BoxedError,
+        #[snafu(implicit)]
+        location: Location,
+    },
+}
+
+pub type Result<T> = std::result::Result<T, Error>;
+
+impl ErrorExt for Error {
+    fn status_code(&self) -> StatusCode {
+        match self {
+            Error::MismatchedSchema { .. } | Error::SerializeEvent { .. } => {
+                StatusCode::InvalidArguments
+            }
+            Error::InsertEvents { .. } | Error::KvBackend { .. } => StatusCode::Internal,
+        }
+    }
+
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+}

src/common/event-recorder/src/lib.rs

@@ -0,0 +1,20 @@
+// 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.
+
+#![feature(duration_constructors)]
+
+pub mod error;
+pub mod recorder;
+
+pub use recorder::*;

src/common/event-recorder/src/recorder.rs

@@ -0,0 +1,668 @@
+// 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::any::Any;
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::sync::Arc;
+use std::time::Duration;
+
+use api::v1::column_data_type_extension::TypeExt;
+use api::v1::value::ValueData;
+use api::v1::{
+    ColumnDataType, ColumnDataTypeExtension, ColumnSchema, JsonTypeExtension, Row,
+    RowInsertRequest, RowInsertRequests, Rows, SemanticType,
+};
+use async_trait::async_trait;
+use backon::{BackoffBuilder, ExponentialBuilder};
+use common_telemetry::{debug, error, info, warn};
+use common_time::timestamp::{TimeUnit, Timestamp};
+use humantime::format_duration;
+use itertools::Itertools;
+use serde::{Deserialize, Serialize};
+use store_api::mito_engine_options::{APPEND_MODE_KEY, TTL_KEY};
+use tokio::sync::mpsc::{channel, Receiver, Sender};
+use tokio::task::JoinHandle;
+use tokio::time::sleep;
+use tokio_util::sync::CancellationToken;
+
+use crate::error::{MismatchedSchemaSnafu, Result};
+
+/// The default table name for storing the events.
+pub const DEFAULT_EVENTS_TABLE_NAME: &str = "events";
+
+/// The column name for the event type.
+pub const EVENTS_TABLE_TYPE_COLUMN_NAME: &str = "type";
+/// The column name for the event payload.
+pub const EVENTS_TABLE_PAYLOAD_COLUMN_NAME: &str = "payload";
+/// The column name for the event timestamp.
+pub const EVENTS_TABLE_TIMESTAMP_COLUMN_NAME: &str = "timestamp";
+
+/// EventRecorderRef is the reference to the event recorder.
+pub type EventRecorderRef = Arc<dyn EventRecorder>;
+
+/// The time interval for flushing batched events to the event handler.
+pub const DEFAULT_FLUSH_INTERVAL_SECONDS: Duration = Duration::from_secs(5);
+/// The default TTL(90 days) for the events table.
+const DEFAULT_EVENTS_TABLE_TTL: Duration = Duration::from_days(90);
+// The capacity of the tokio channel for transmitting events to background processor.
+const DEFAULT_CHANNEL_SIZE: usize = 2048;
+// The size of the buffer for batching events before flushing to event handler.
+const DEFAULT_BUFFER_SIZE: usize = 100;
+// The maximum number of retry attempts when event handler processing fails.
+const DEFAULT_MAX_RETRY_TIMES: u64 = 3;
+
+/// Event trait defines the interface for events that can be recorded and persisted as the system table.
+/// By default, the event will be persisted as the system table with the following schema:
+///
+/// - `type`: the type of the event.
+/// - `payload`: the JSON bytes of the event.
+/// - `timestamp`: the timestamp of the event.
+///
+/// The event can also add the extra schema and row to the event by overriding the `extra_schema` and `extra_row` methods.
+pub trait Event: Send + Sync + Debug {
+    /// Returns the table name of the event.
+    fn table_name(&self) -> &str {
+        DEFAULT_EVENTS_TABLE_NAME
+    }
+
+    /// Returns the type of the event.
+    fn event_type(&self) -> &str;
+
+    /// Returns the timestamp of the event. Default to the current time.
+    fn timestamp(&self) -> Timestamp {
+        Timestamp::current_time(TimeUnit::Nanosecond)
+    }
+
+    /// Returns the JSON bytes of the event as the payload. It will use JSON type to store the payload.
+    fn json_payload(&self) -> Result<String> {
+        Ok("".to_string())
+    }
+
+    /// Add the extra schema to the event with the default schema.
+    fn extra_schema(&self) -> Vec<ColumnSchema> {
+        vec![]
+    }
+
+    /// Add the extra row to the event with the default row.
+    fn extra_row(&self) -> Result<Row> {
+        Ok(Row { values: vec![] })
+    }
+
+    /// Returns the event as any type.
+    fn as_any(&self) -> &dyn Any;
+}
+
+/// Eventable trait defines the interface for objects that can be converted to [Event].
+pub trait Eventable: Send + Sync + Debug {
+    /// Converts the object to an [Event].
+    fn to_event(&self) -> Option<Box<dyn Event>> {
+        None
+    }
+}
+
+/// Groups events by its `event_type`.
+#[allow(clippy::borrowed_box)]
+pub fn group_events_by_type(events: &[Box<dyn Event>]) -> HashMap<&str, Vec<&Box<dyn Event>>> {
+    events
+        .iter()
+        .into_grouping_map_by(|event| event.event_type())
+        .collect()
+}
+
+/// Builds the row inserts request for the events that will be persisted to the events table. The `events` should have the same event type, or it will return an error.
+#[allow(clippy::borrowed_box)]
+pub fn build_row_inserts_request(events: &[&Box<dyn Event>]) -> Result<RowInsertRequests> {
+    // Ensure all the events are the same type.
+    validate_events(events)?;
+
+    // We already validated the events, so it's safe to get the first event to build the schema for the RowInsertRequest.
+    let event = &events[0];
+    let mut schema: Vec<ColumnSchema> = Vec::with_capacity(3 + event.extra_schema().len());
+    schema.extend(vec![
+        ColumnSchema {
+            column_name: EVENTS_TABLE_TYPE_COLUMN_NAME.to_string(),
+            datatype: ColumnDataType::String.into(),
+            semantic_type: SemanticType::Tag.into(),
+            ..Default::default()
+        },
+        ColumnSchema {
+            column_name: EVENTS_TABLE_PAYLOAD_COLUMN_NAME.to_string(),
+            datatype: ColumnDataType::Binary as i32,
+            semantic_type: SemanticType::Field as i32,
+            datatype_extension: Some(ColumnDataTypeExtension {
+                type_ext: Some(TypeExt::JsonType(JsonTypeExtension::JsonBinary.into())),
+            }),
+            ..Default::default()
+        },
+        ColumnSchema {
+            column_name: EVENTS_TABLE_TIMESTAMP_COLUMN_NAME.to_string(),
+            datatype: ColumnDataType::TimestampNanosecond.into(),
+            semantic_type: SemanticType::Timestamp.into(),
+            ..Default::default()
+        },
+    ]);
+    schema.extend(event.extra_schema());
+
+    let mut rows: Vec<Row> = Vec::with_capacity(events.len());
+    for event in events {
+        let extra_row = event.extra_row()?;
+        let mut values = Vec::with_capacity(3 + extra_row.values.len());
+        values.extend([
+            ValueData::StringValue(event.event_type().to_string()).into(),
+            ValueData::BinaryValue(event.json_payload()?.into_bytes()).into(),
+            ValueData::TimestampNanosecondValue(event.timestamp().value()).into(),
+        ]);
+        values.extend(extra_row.values);
+        rows.push(Row { values });
+    }
+
+    Ok(RowInsertRequests {
+        inserts: vec![RowInsertRequest {
+            table_name: event.table_name().to_string(),
+            rows: Some(Rows { schema, rows }),
+        }],
+    })
+}
+
+// Ensure the events with the same event type have the same extra schema.
+#[allow(clippy::borrowed_box)]
+fn validate_events(events: &[&Box<dyn Event>]) -> Result<()> {
+    // It's safe to get the first event because the events are already grouped by the event type.
+    let extra_schema = events[0].extra_schema();
+    for event in events {
+        if event.extra_schema() != extra_schema {
+            MismatchedSchemaSnafu {
+                expected: extra_schema.clone(),
+                actual: event.extra_schema(),
+            }
+            .fail()?;
+        }
+    }
+    Ok(())
+}
+
+/// EventRecorder trait defines the interface for recording events.
+pub trait EventRecorder: Send + Sync + Debug + 'static {
+    /// Records an event for persistence and processing by [EventHandler].
+    fn record(&self, event: Box<dyn Event>);
+
+    /// Cancels the event recorder.
+    fn close(&self);
+}
+
+/// EventHandlerOptions is the options for the event handler.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct EventHandlerOptions {
+    /// TTL for the events table that will be used to store the events.
+    pub ttl: Duration,
+    /// Append mode for the events table that will be used to store the events.
+    pub append_mode: bool,
+}
+
+impl Default for EventHandlerOptions {
+    fn default() -> Self {
+        Self {
+            ttl: DEFAULT_EVENTS_TABLE_TTL,
+            append_mode: true,
+        }
+    }
+}
+
+impl EventHandlerOptions {
+    /// Converts the options to the hints for the insert operation.
+    pub fn to_hints(&self) -> Vec<(&str, String)> {
+        vec![
+            (TTL_KEY, format_duration(self.ttl).to_string()),
+            (APPEND_MODE_KEY, self.append_mode.to_string()),
+        ]
+    }
+}
+
+/// EventHandler trait defines the interface for how to handle the event.
+#[async_trait]
+pub trait EventHandler: Send + Sync + 'static {
+    /// Processes and handles incoming events. The [DefaultEventHandlerImpl] implementation forwards events to frontend instances for persistence.
+    /// We use `&[Box<dyn Event>]` to avoid consuming the events, so the caller can buffer the events and retry if the handler fails.
+    async fn handle(&self, events: &[Box<dyn Event>]) -> Result<()>;
+}
+
+/// Configuration options for the event recorder.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct EventRecorderOptions {
+    /// TTL for the events table that will be used to store the events.
+    #[serde(with = "humantime_serde")]
+    pub ttl: Duration,
+}
+
+impl Default for EventRecorderOptions {
+    fn default() -> Self {
+        Self {
+            ttl: DEFAULT_EVENTS_TABLE_TTL,
+        }
+    }
+}
+
+/// Implementation of [EventRecorder] that records the events and processes them in the background by the [EventHandler].
+#[derive(Debug)]
+pub struct EventRecorderImpl {
+    // The channel to send the events to the background processor.
+    tx: Sender<Box<dyn Event>>,
+    // The cancel token to cancel the background processor.
+    cancel_token: CancellationToken,
+    // The background processor to process the events.
+    handle: Option<JoinHandle<()>>,
+}
+
+impl EventRecorderImpl {
+    pub fn new(event_handler: Box<dyn EventHandler>) -> Self {
+        let (tx, rx) = channel(DEFAULT_CHANNEL_SIZE);
+        let cancel_token = CancellationToken::new();
+
+        let mut recorder = Self {
+            tx,
+            handle: None,
+            cancel_token: cancel_token.clone(),
+        };
+
+        let processor = EventProcessor::new(
+            rx,
+            event_handler,
+            DEFAULT_FLUSH_INTERVAL_SECONDS,
+            DEFAULT_MAX_RETRY_TIMES,
+        )
+        .with_cancel_token(cancel_token);
+
+        // Spawn a background task to process the events.
+        let handle = tokio::spawn(async move {
+            processor.process(DEFAULT_BUFFER_SIZE).await;
+        });
+
+        recorder.handle = Some(handle);
+
+        recorder
+    }
+}
+
+impl EventRecorder for EventRecorderImpl {
+    // Accepts an event and send it to the background handler.
+    fn record(&self, event: Box<dyn Event>) {
+        if let Err(e) = self.tx.try_send(event) {
+            error!("Failed to send event to the background processor: {}", e);
+        }
+    }
+
+    // Closes the event recorder. It will stop the background processor and flush the buffer.
+    fn close(&self) {
+        self.cancel_token.cancel();
+    }
+}
+
+impl Drop for EventRecorderImpl {
+    fn drop(&mut self) {
+        if let Some(handle) = self.handle.take() {
+            handle.abort();
+            info!("Aborted the background processor in event recorder");
+        }
+    }
+}
+
+struct EventProcessor {
+    rx: Receiver<Box<dyn Event>>,
+    event_handler: Box<dyn EventHandler>,
+    max_retry_times: u64,
+    process_interval: Duration,
+    cancel_token: CancellationToken,
+}
+
+impl EventProcessor {
+    fn new(
+        rx: Receiver<Box<dyn Event>>,
+        event_handler: Box<dyn EventHandler>,
+        process_interval: Duration,
+        max_retry_times: u64,
+    ) -> Self {
+        Self {
+            rx,
+            event_handler,
+            max_retry_times,
+            process_interval,
+            cancel_token: CancellationToken::new(),
+        }
+    }
+
+    fn with_cancel_token(mut self, cancel_token: CancellationToken) -> Self {
+        self.cancel_token = cancel_token;
+        self
+    }
+
+    async fn process(mut self, buffer_size: usize) {
+        info!("Start the background processor in event recorder to handle the received events.");
+
+        let mut buffer = Vec::with_capacity(buffer_size);
+        let mut interval = tokio::time::interval(self.process_interval);
+
+        loop {
+            tokio::select! {
+                maybe_event = self.rx.recv() => {
+                    if let Some(maybe_event) = maybe_event {
+                        debug!("Received event: {:?}", maybe_event);
+
+                        if buffer.len() >= buffer_size {
+                            debug!(
+                                "Flushing events to the event handler because the buffer is full with {} events",
+                                buffer.len()
+                            );
+                            self.flush_events_to_handler(&mut buffer).await;
+                        }
+
+                        // Push the event to the buffer, the buffer will be flushed when the interval is triggered or received a closed signal.
+                        buffer.push(maybe_event);
+                    } else {
+                        // When received a closed signal, flush the buffer and exit the loop.
+                        self.flush_events_to_handler(&mut buffer).await;
+                        break;
+                    }
+                }
+                // Cancel the processor through the cancel token.
+                _ = self.cancel_token.cancelled() => {
+                    warn!("Received a cancel signal, flushing the buffer and exiting the loop");
+                    self.flush_events_to_handler(&mut buffer).await;
+                    break;
+                }
+                // When the interval is triggered, flush the buffer and send the events to the event handler.
+                _ = interval.tick() => {
+                    self.flush_events_to_handler(&mut buffer).await;
+                }
+            }
+        }
+    }
+
+    // NOTE: While we implement a retry mechanism for failed event handling, there is no guarantee that all events will be processed successfully.
+    async fn flush_events_to_handler(&self, buffer: &mut Vec<Box<dyn Event>>) {
+        if !buffer.is_empty() {
+            debug!("Flushing {} events to the event handler", buffer.len());
+
+            let mut backoff = ExponentialBuilder::default()
+                .with_min_delay(Duration::from_millis(
+                    DEFAULT_FLUSH_INTERVAL_SECONDS.as_millis() as u64 / self.max_retry_times.max(1),
+                ))
+                .with_max_delay(Duration::from_millis(
+                    DEFAULT_FLUSH_INTERVAL_SECONDS.as_millis() as u64,
+                ))
+                .with_max_times(self.max_retry_times as usize)
+                .build();
+
+            loop {
+                match self.event_handler.handle(buffer).await {
+                    Ok(()) => {
+                        debug!("Successfully handled {} events", buffer.len());
+                        break;
+                    }
+                    Err(e) => {
+                        if let Some(d) = backoff.next() {
+                            warn!(e; "Failed to handle events, retrying...");
+                            sleep(d).await;
+                            continue;
+                        } else {
+                            warn!(
+                                e; "Failed to handle events after {} retries",
+                                self.max_retry_times
+                            );
+                            break;
+                        }
+                    }
+                }
+            }
+        }
+
+        // Clear the buffer to prevent unbounded memory growth, regardless of whether event processing succeeded or failed.
+        buffer.clear();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[derive(Debug)]
+    struct TestEvent {}
+
+    impl Event for TestEvent {
+        fn event_type(&self) -> &str {
+            "test_event"
+        }
+
+        fn json_payload(&self) -> Result<String> {
+            Ok("{\"procedure_id\": \"1234567890\"}".to_string())
+        }
+
+        fn as_any(&self) -> &dyn Any {
+            self
+        }
+    }
+
+    struct TestEventHandlerImpl {}
+
+    #[async_trait]
+    impl EventHandler for TestEventHandlerImpl {
+        async fn handle(&self, events: &[Box<dyn Event>]) -> Result<()> {
+            let event = events
+                .first()
+                .unwrap()
+                .as_any()
+                .downcast_ref::<TestEvent>()
+                .unwrap();
+            assert_eq!(
+                event.json_payload().unwrap(),
+                "{\"procedure_id\": \"1234567890\"}"
+            );
+            assert_eq!(event.event_type(), "test_event");
+            Ok(())
+        }
+    }
+
+    #[tokio::test]
+    async fn test_event_recorder() {
+        let mut event_recorder = EventRecorderImpl::new(Box::new(TestEventHandlerImpl {}));
+        event_recorder.record(Box::new(TestEvent {}));
+
+        // Sleep for a while to let the event be sent to the event handler.
+        sleep(Duration::from_millis(500)).await;
+
+        // Close the event recorder to flush the buffer.
+        event_recorder.close();
+
+        // Sleep for a while to let the background task process the event.
+        sleep(Duration::from_millis(500)).await;
+
+        if let Some(handle) = event_recorder.handle.take() {
+            assert!(handle.await.is_ok());
+        }
+    }
+
+    struct TestEventHandlerImplShouldPanic {}
+
+    #[async_trait]
+    impl EventHandler for TestEventHandlerImplShouldPanic {
+        async fn handle(&self, events: &[Box<dyn Event>]) -> Result<()> {
+            let event = events
+                .first()
+                .unwrap()
+                .as_any()
+                .downcast_ref::<TestEvent>()
+                .unwrap();
+
+            // Set the incorrect payload and event type to trigger the panic.
+            assert_eq!(
+                event.json_payload().unwrap(),
+                "{\"procedure_id\": \"should_panic\"}"
+            );
+            assert_eq!(event.event_type(), "should_panic");
+            Ok(())
+        }
+    }
+
+    #[tokio::test]
+    async fn test_event_recorder_should_panic() {
+        let mut event_recorder =
+            EventRecorderImpl::new(Box::new(TestEventHandlerImplShouldPanic {}));
+
+        event_recorder.record(Box::new(TestEvent {}));
+
+        // Sleep for a while to let the event be sent to the event handler.
+        sleep(Duration::from_millis(500)).await;
+
+        // Close the event recorder to flush the buffer.
+        event_recorder.close();
+
+        // Sleep for a while to let the background task process the event.
+        sleep(Duration::from_millis(500)).await;
+
+        if let Some(handle) = event_recorder.handle.take() {
+            assert!(handle.await.unwrap_err().is_panic());
+        }
+    }
+
+    #[derive(Debug)]
+    struct TestEventA {}
+
+    impl Event for TestEventA {
+        fn event_type(&self) -> &str {
+            "A"
+        }
+
+        fn as_any(&self) -> &dyn Any {
+            self
+        }
+    }
+
+    #[derive(Debug)]
+    struct TestEventB {}
+
+    impl Event for TestEventB {
+        fn table_name(&self) -> &str {
+            "table_B"
+        }
+
+        fn event_type(&self) -> &str {
+            "B"
+        }
+
+        fn as_any(&self) -> &dyn Any {
+            self
+        }
+    }
+
+    #[derive(Debug)]
+    struct TestEventC {}
+
+    impl Event for TestEventC {
+        fn table_name(&self) -> &str {
+            "table_C"
+        }
+
+        fn event_type(&self) -> &str {
+            "C"
+        }
+
+        fn as_any(&self) -> &dyn Any {
+            self
+        }
+    }
+
+    #[test]
+    fn test_group_events_by_type() {
+        let events: Vec<Box<dyn Event>> = vec![
+            Box::new(TestEventA {}),
+            Box::new(TestEventB {}),
+            Box::new(TestEventA {}),
+            Box::new(TestEventC {}),
+            Box::new(TestEventB {}),
+            Box::new(TestEventC {}),
+            Box::new(TestEventA {}),
+        ];
+
+        let event_groups = group_events_by_type(&events);
+        assert_eq!(event_groups.len(), 3);
+        assert_eq!(event_groups.get("A").unwrap().len(), 3);
+        assert_eq!(event_groups.get("B").unwrap().len(), 2);
+        assert_eq!(event_groups.get("C").unwrap().len(), 2);
+    }
+
+    #[test]
+    fn test_build_row_inserts_request() {
+        let events: Vec<Box<dyn Event>> = vec![
+            Box::new(TestEventA {}),
+            Box::new(TestEventB {}),
+            Box::new(TestEventA {}),
+            Box::new(TestEventC {}),
+            Box::new(TestEventB {}),
+            Box::new(TestEventC {}),
+            Box::new(TestEventA {}),
+        ];
+
+        let event_groups = group_events_by_type(&events);
+        assert_eq!(event_groups.len(), 3);
+        assert_eq!(event_groups.get("A").unwrap().len(), 3);
+        assert_eq!(event_groups.get("B").unwrap().len(), 2);
+        assert_eq!(event_groups.get("C").unwrap().len(), 2);
+
+        for (event_type, events) in event_groups {
+            let row_inserts_request = build_row_inserts_request(&events).unwrap();
+            if event_type == "A" {
+                assert_eq!(row_inserts_request.inserts.len(), 1);
+                assert_eq!(
+                    row_inserts_request.inserts[0].table_name,
+                    DEFAULT_EVENTS_TABLE_NAME
+                );
+                assert_eq!(
+                    row_inserts_request.inserts[0]
+                        .rows
+                        .as_ref()
+                        .unwrap()
+                        .rows
+                        .len(),
+                    3
+                );
+            } else if event_type == "B" {
+                assert_eq!(row_inserts_request.inserts.len(), 1);
+                assert_eq!(row_inserts_request.inserts[0].table_name, "table_B");
+                assert_eq!(
+                    row_inserts_request.inserts[0]
+                        .rows
+                        .as_ref()
+                        .unwrap()
+                        .rows
+                        .len(),
+                    2
+                );
+            } else if event_type == "C" {
+                assert_eq!(row_inserts_request.inserts.len(), 1);
+                assert_eq!(row_inserts_request.inserts[0].table_name, "table_C");
+                assert_eq!(
+                    row_inserts_request.inserts[0]
+                        .rows
+                        .as_ref()
+                        .unwrap()
+                        .rows
+                        .len(),
+                    2
+                );
+            } else {
+                panic!("Unexpected event type: {}", event_type);
+            }
+        }
+    }
+}

src/common/frontend/Cargo.toml

@@ -5,13 +5,19 @@ edition.workspace = true
 license.workspace = true
 
 [dependencies]
+api.workspace = true
 async-trait.workspace = true
 common-error.workspace = true
+common-event-recorder.workspace = true
 common-grpc.workspace = true
 common-macro.workspace = true
 common-meta.workspace = true
+common-time.workspace = true
 greptime-proto.workspace = true
+humantime.workspace = true
 meta-client.workspace = true
+serde.workspace = true
+session.workspace = true
 snafu.workspace = true
 tonic.workspace = true
 

src/common/frontend/src/lib.rs

@@ -19,6 +19,7 @@ use snafu::OptionExt;
 
 pub mod error;
 pub mod selector;
+pub mod slow_query_event;
 
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct DisplayProcessId {

src/common/frontend/src/slow_query_event.rs

@@ -0,0 +1,132 @@
+// 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::any::Any;
+
+use api::v1::value::ValueData;
+use api::v1::{ColumnDataType, ColumnSchema, Row, SemanticType};
+use common_event_recorder::error::Result;
+use common_event_recorder::Event;
+use serde::Serialize;
+
+pub const SLOW_QUERY_TABLE_NAME: &str = "slow_queries";
+pub const SLOW_QUERY_TABLE_COST_COLUMN_NAME: &str = "cost";
+pub const SLOW_QUERY_TABLE_THRESHOLD_COLUMN_NAME: &str = "threshold";
+pub const SLOW_QUERY_TABLE_QUERY_COLUMN_NAME: &str = "query";
+pub const SLOW_QUERY_TABLE_TIMESTAMP_COLUMN_NAME: &str = "timestamp";
+pub const SLOW_QUERY_TABLE_IS_PROMQL_COLUMN_NAME: &str = "is_promql";
+pub const SLOW_QUERY_TABLE_PROMQL_START_COLUMN_NAME: &str = "promql_start";
+pub const SLOW_QUERY_TABLE_PROMQL_END_COLUMN_NAME: &str = "promql_end";
+pub const SLOW_QUERY_TABLE_PROMQL_RANGE_COLUMN_NAME: &str = "promql_range";
+pub const SLOW_QUERY_TABLE_PROMQL_STEP_COLUMN_NAME: &str = "promql_step";
+pub const SLOW_QUERY_EVENT_TYPE: &str = "slow_query";
+
+/// SlowQueryEvent is the event of slow query.
+#[derive(Debug, Serialize)]
+pub struct SlowQueryEvent {
+    pub cost: u64,
+    pub threshold: u64,
+    pub query: String,
+    pub is_promql: bool,
+    pub promql_range: Option<u64>,
+    pub promql_step: Option<u64>,
+    pub promql_start: Option<i64>,
+    pub promql_end: Option<i64>,
+}
+
+impl Event for SlowQueryEvent {
+    fn table_name(&self) -> &str {
+        SLOW_QUERY_TABLE_NAME
+    }
+
+    fn event_type(&self) -> &str {
+        SLOW_QUERY_EVENT_TYPE
+    }
+
+    fn extra_schema(&self) -> Vec<ColumnSchema> {
+        vec![
+            ColumnSchema {
+                column_name: SLOW_QUERY_TABLE_COST_COLUMN_NAME.to_string(),
+                datatype: ColumnDataType::Uint64.into(),
+                semantic_type: SemanticType::Field.into(),
+                ..Default::default()
+            },
+            ColumnSchema {
+                column_name: SLOW_QUERY_TABLE_THRESHOLD_COLUMN_NAME.to_string(),
+                datatype: ColumnDataType::Uint64.into(),
+                semantic_type: SemanticType::Field.into(),
+                ..Default::default()
+            },
+            ColumnSchema {
+                column_name: SLOW_QUERY_TABLE_QUERY_COLUMN_NAME.to_string(),
+                datatype: ColumnDataType::String.into(),
+                semantic_type: SemanticType::Field.into(),
+                ..Default::default()
+            },
+            ColumnSchema {
+                column_name: SLOW_QUERY_TABLE_IS_PROMQL_COLUMN_NAME.to_string(),
+                datatype: ColumnDataType::Boolean.into(),
+                semantic_type: SemanticType::Field.into(),
+                ..Default::default()
+            },
+            ColumnSchema {
+                column_name: SLOW_QUERY_TABLE_PROMQL_RANGE_COLUMN_NAME.to_string(),
+                datatype: ColumnDataType::Uint64.into(),
+                semantic_type: SemanticType::Field.into(),
+                ..Default::default()
+            },
+            ColumnSchema {
+                column_name: SLOW_QUERY_TABLE_PROMQL_STEP_COLUMN_NAME.to_string(),
+                datatype: ColumnDataType::Uint64.into(),
+                semantic_type: SemanticType::Field.into(),
+                ..Default::default()
+            },
+            ColumnSchema {
+                column_name: SLOW_QUERY_TABLE_PROMQL_START_COLUMN_NAME.to_string(),
+                datatype: ColumnDataType::TimestampMillisecond.into(),
+                semantic_type: SemanticType::Field.into(),
+                ..Default::default()
+            },
+            ColumnSchema {
+                column_name: SLOW_QUERY_TABLE_PROMQL_END_COLUMN_NAME.to_string(),
+                datatype: ColumnDataType::TimestampMillisecond.into(),
+                semantic_type: SemanticType::Field.into(),
+                ..Default::default()
+            },
+        ]
+    }
+
+    fn extra_row(&self) -> Result<Row> {
+        Ok(Row {
+            values: vec![
+                ValueData::U64Value(self.cost).into(),
+                ValueData::U64Value(self.threshold).into(),
+                ValueData::StringValue(self.query.to_string()).into(),
+                ValueData::BoolValue(self.is_promql).into(),
+                ValueData::U64Value(self.promql_range.unwrap_or(0)).into(),
+                ValueData::U64Value(self.promql_step.unwrap_or(0)).into(),
+                ValueData::TimestampMillisecondValue(self.promql_start.unwrap_or(0)).into(),
+                ValueData::TimestampMillisecondValue(self.promql_end.unwrap_or(0)).into(),
+            ],
+        })
+    }
+
+    fn json_payload(&self) -> Result<String> {
+        Ok("".to_string())
+    }
+
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+}

src/common/function/Cargo.toml

@@ -16,6 +16,8 @@ geo = ["geohash", "h3o", "s2", "wkt", "geo-types", "dep:geo"]
 ahash.workspace = true
 api.workspace = true
 arc-swap = "1.0"
+arrow.workspace = true
+arrow-schema.workspace = true
 async-trait.workspace = true
 bincode = "1.3"
 catalog.workspace = true
@@ -34,6 +36,7 @@ datafusion.workspace = true
 datafusion-common.workspace = true
 datafusion-expr.workspace = true
 datafusion-functions-aggregate-common.workspace = true
+datafusion-physical-expr.workspace = true
 datatypes.workspace = true
 derive_more = { version = "1", default-features = false, features = ["display"] }
 geo = { version = "0.29", optional = true }
@@ -62,5 +65,7 @@ wkt = { version = "0.11", optional = true }
 
 [dev-dependencies]
 approx = "0.5"
+futures.workspace = true
+pretty_assertions.workspace = true
 serde = { version = "1.0", features = ["derive"] }
 tokio.workspace = true