refactor!: remove not working metasrv cli option

Signed-off-by: luofucong <luofc@foxmail.com>

.github/workflows/check-git-deps.yml

@@ -0,0 +1,154 @@
+name: Check Git Dependencies on Main Branch
+
+on:
+    pull_request:
+        branches: [main]
+        paths:
+            - 'Cargo.toml'
+    push:
+        branches: [main]
+        paths:
+            - 'Cargo.toml'
+
+jobs:
+    check-git-deps:
+        runs-on: ubuntu-latest
+
+        steps:
+            - name: Checkout repository
+              uses: actions/checkout@v6
+
+            - name: Check git dependencies
+              env:
+                  WHITELIST_DEPS: "greptime-proto,meter-core,meter-macros"
+              run: |
+                  #!/bin/bash
+                  set -e
+
+                  echo "Checking whitelisted git dependencies..."
+
+                  # Function to check if a commit is on main branch
+                  check_commit_on_main() {
+                      local repo_url="$1"
+                      local commit="$2"
+                      local repo_name=$(basename "$repo_url" .git)
+
+                      echo "Checking $repo_name"
+                      echo "Repo: $repo_url"
+                      echo "Commit: $commit"
+
+                      # Create a temporary directory for cloning
+                      local temp_dir=$(mktemp -d)
+
+                      # Clone the repository
+                      if git clone "$repo_url" "$temp_dir" 2>/dev/null; then
+                          cd "$temp_dir"
+
+                          # Try to determine the main branch name
+                          local main_branch="main"
+                          if ! git rev-parse --verify origin/main >/dev/null 2>&1; then
+                              if git rev-parse --verify origin/master >/dev/null 2>&1; then
+                                  main_branch="master"
+                              else
+                                  # Try to get the default branch
+                                  main_branch=$(git symbolic-ref refs/remotes/origin/HEAD | sed 's@^refs/remotes/origin/@@')
+                              fi
+                          fi
+
+                          echo "Main branch: $main_branch"
+
+                          # Check if commit exists
+                          if git cat-file -e "$commit" 2>/dev/null; then
+                              # Check if commit is on main branch
+                              if git merge-base --is-ancestor "$commit" "origin/$main_branch" 2>/dev/null; then
+                                  echo "PASS: Commit $commit is on $main_branch branch"
+                                  cd - >/dev/null
+                                  rm -rf "$temp_dir"
+                                  return 0
+                              else
+                                  echo "FAIL: Commit $commit is NOT on $main_branch branch"
+
+                                  # Try to find which branch contains this commit
+                                  local branch_name=$(git branch -r --contains "$commit" 2>/dev/null | head -1 | sed 's/^[[:space:]]*origin\///' | sed 's/[[:space:]]*$//')
+                                  if [[ -n "$branch_name" ]]; then
+                                      echo "Found on branch: $branch_name"
+                                  fi
+                                  cd - >/dev/null
+                                  rm -rf "$temp_dir"
+                                  return 1
+                              fi
+                          else
+                              echo "FAIL: Commit $commit not found in repository"
+                              cd - >/dev/null
+                              rm -rf "$temp_dir"
+                              return 1
+                          fi
+                      else
+                          echo "FAIL: Failed to clone $repo_url"
+                          rm -rf "$temp_dir"
+                          return 1
+                      fi
+                  }
+
+                  # Extract whitelisted git dependencies from Cargo.toml
+                  echo "Extracting git dependencies from Cargo.toml..."
+
+                  # Create temporary array to store dependencies
+                  declare -a deps=()
+
+                  # Build awk pattern from whitelist
+                  IFS=',' read -ra WHITELIST <<< "$WHITELIST_DEPS"
+                  awk_pattern=""
+                  for dep in "${WHITELIST[@]}"; do
+                      if [[ -n "$awk_pattern" ]]; then
+                          awk_pattern="$awk_pattern|"
+                      fi
+                      awk_pattern="$awk_pattern$dep"
+                  done
+
+                  # Extract whitelisted dependencies
+                  while IFS= read -r line; do
+                      if [[ -n "$line" ]]; then
+                          deps+=("$line")
+                      fi
+                  done < <(awk -v pattern="$awk_pattern" '
+                  $0 ~ pattern ".*git = \"https:/" {
+                      match($0, /git = "([^"]+)"/, arr)
+                      git_url = arr[1]
+                      if (match($0, /rev = "([^"]+)"/, rev_arr)) {
+                          rev = rev_arr[1]
+                          print git_url " " rev
+                      } else {
+                          # Check next line for rev
+                          getline
+                          if (match($0, /rev = "([^"]+)"/, rev_arr)) {
+                              rev = rev_arr[1]
+                              print git_url " " rev
+                          }
+                      }
+                  }
+                  ' Cargo.toml)
+
+                  echo "Found ${#deps[@]} dependencies to check:"
+                  for dep in "${deps[@]}"; do
+                      echo "  $dep"
+                  done
+
+                  failed=0
+
+                  for dep in "${deps[@]}"; do
+                      read -r repo_url commit <<< "$dep"
+                      if ! check_commit_on_main "$repo_url" "$commit"; then
+                          failed=1
+                      fi
+                  done
+
+                  echo "Check completed."
+
+                  if [[ $failed -eq 1 ]]; then
+                      echo "ERROR: Some git dependencies are not on their main branches!"
+                      echo "Please update the commits to point to main branch commits."
+                      exit 1
+                  else
+                      echo "SUCCESS: All git dependencies are on their main branches!"
+                  fi

Cargo.lock

@@ -1786,6 +1786,7 @@ dependencies = [
  "common-recordbatch",
  "common-runtime",
  "common-telemetry",
+ "common-test-util",
  "common-time",
  "common-version",
  "common-wal",
@@ -4633,8 +4634,9 @@ dependencies = [
 
 [[package]]
 name = "etcd-client"
-version = "0.15.0"
-source = "git+https://github.com/GreptimeTeam/etcd-client?rev=f62df834f0cffda355eba96691fe1a9a332b75a7#f62df834f0cffda355eba96691fe1a9a332b75a7"
+version = "0.16.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "88365f1a5671eb2f7fc240adb216786bc6494b38ce15f1d26ad6eaa303d5e822"
 dependencies = [
  "http 1.3.1",
  "prost 0.13.5",
@@ -5459,7 +5461,7 @@ dependencies = [
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=0423fa30203187c75e2937a668df1da699c8b96c#0423fa30203187c75e2937a668df1da699c8b96c"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=173efe5ec62722089db7c531c0b0d470a072b915#173efe5ec62722089db7c531c0b0d470a072b915"
 dependencies = [
  "prost 0.13.5",
  "prost-types 0.13.5",

Cargo.toml

@@ -143,14 +143,14 @@ derive_builder = "0.20"
 derive_more = { version = "2.1", features = ["full"] }
 dotenv = "0.15"
 either = "1.15"
-etcd-client = { git = "https://github.com/GreptimeTeam/etcd-client", rev = "f62df834f0cffda355eba96691fe1a9a332b75a7", features = [
+etcd-client = { version = "0.16.1", features = [
     "tls",
     "tls-roots",
 ] }
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "0423fa30203187c75e2937a668df1da699c8b96c" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "173efe5ec62722089db7c531c0b0d470a072b915" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"

config/config.md

@@ -83,6 +83,8 @@
 | `wal.sync_period` | String | `10s` | Duration for fsyncing log files.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.recovery_parallelism` | Integer | `2` | Parallelism during WAL recovery. |
 | `wal.broker_endpoints` | Array | -- | The Kafka broker endpoints.<br/>**It's only used when the provider is `kafka`**. |
+| `wal.connect_timeout` | String | `3s` | The connect timeout for kafka client.<br/>**It's only used when the provider is `kafka`**. |
+| `wal.timeout` | String | `3s` | The timeout for kafka client.<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)` |
 | `wal.num_topics` | Integer | `64` | Number of topics.<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`**. |
@@ -347,11 +349,11 @@
 | `meta_schema_name` | String | `greptime_schema` | Optional PostgreSQL schema for metadata table and election table name qualification.<br/>When PostgreSQL public schema is not writable (e.g., PostgreSQL 15+ with restricted public),<br/>set this to a writable schema. GreptimeDB will use `meta_schema_name`.`meta_table_name`.<br/>GreptimeDB will NOT create the schema automatically; please ensure it exists or the user has permission.<br/>**Only used when backend is `postgres_store`.** |
 | `meta_election_lock_id` | Integer | `1` | Advisory lock id in PostgreSQL for election. Effect when using PostgreSQL as kvbackend<br/>Only used when backend is `postgres_store`. |
 | `selector` | String | `round_robin` | Datanode selector type.<br/>- `round_robin` (default value)<br/>- `lease_based`<br/>- `load_based`<br/>For details, please see "https://docs.greptime.com/developer-guide/metasrv/selector". |
-| `use_memory_store` | Bool | `false` | Store data in memory. |
 | `enable_region_failover` | Bool | `false` | Whether to enable region failover.<br/>This feature is only available on GreptimeDB running on cluster mode and<br/>- Using Remote WAL<br/>- Using shared storage (e.g., s3). |
 | `region_failure_detector_initialization_delay` | String | `10m` | The delay before starting region failure detection.<br/>This delay helps prevent Metasrv from triggering unnecessary region failovers before all Datanodes are fully started.<br/>Especially useful when the cluster is not deployed with GreptimeDB Operator and maintenance mode is not enabled. |
 | `allow_region_failover_on_local_wal` | Bool | `false` | Whether to allow region failover on local WAL.<br/>**This option is not recommended to be set to true, because it may lead to data loss during failover.** |
 | `node_max_idle_time` | String | `24hours` | Max allowed idle time before removing node info from metasrv memory. |
+| `heartbeat_interval` | String | `3s` | Base heartbeat interval for calculating distributed time constants.<br/>The frontend heartbeat interval is 6 times of the base heartbeat interval.<br/>The flownode/datanode heartbeat interval is 1 times of the base heartbeat interval.<br/>e.g., If the base heartbeat interval is 3s, the frontend heartbeat interval is 18s, the flownode/datanode heartbeat interval is 3s.<br/>If you change this value, you need to change the heartbeat interval of the flownode/frontend/datanode accordingly. |
 | `enable_telemetry` | Bool | `true` | Whether to enable greptimedb telemetry. Enabled by default. |
 | `runtime` | -- | -- | The runtime options. |
 | `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
@@ -361,12 +363,18 @@
 | `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_client` | -- | -- | The backend client options.<br/>Currently, only applicable when using etcd as the metadata store. |
+| `backend_client.keep_alive_timeout` | String | `3s` | The keep alive timeout for backend client. |
+| `backend_client.keep_alive_interval` | String | `10s` | The keep alive interval for backend client. |
+| `backend_client.connect_timeout` | String | `3s` | The connect timeout for backend client. |
 | `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`. |
 | `grpc.runtime_size` | Integer | `8` | The number of server worker threads. |
 | `grpc.max_recv_message_size` | String | `512MB` | The maximum receive message size for gRPC server. |
 | `grpc.max_send_message_size` | String | `512MB` | The maximum send message size for gRPC server. |
+| `grpc.http2_keep_alive_interval` | String | `10s` | The server side HTTP/2 keep-alive interval |
+| `grpc.http2_keep_alive_timeout` | String | `3s` | The server side HTTP/2 keep-alive timeout. |
 | `http` | -- | -- | The HTTP server options. |
 | `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. |
 | `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout. |
@@ -476,6 +484,8 @@
 | `wal.sync_period` | String | `10s` | Duration for fsyncing log files.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.recovery_parallelism` | Integer | `2` | Parallelism during WAL recovery. |
 | `wal.broker_endpoints` | Array | -- | The Kafka broker endpoints.<br/>**It's only used when the provider is `kafka`**. |
+| `wal.connect_timeout` | String | `3s` | The connect timeout for kafka client.<br/>**It's only used when the provider is `kafka`**. |
+| `wal.timeout` | String | `3s` | The timeout for kafka client.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.max_batch_bytes` | String | `1MB` | The max size of a single producer batch.<br/>Warning: Kafka has a default limit of 1MB per message in a topic.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.consumer_wait_timeout` | String | `100ms` | The consumer wait timeout.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.create_index` | Bool | `true` | Whether to enable WAL index creation.<br/>**It's only used when the provider is `kafka`**. |

config/datanode.example.toml

@@ -169,6 +169,14 @@ recovery_parallelism = 2
 ## **It's only used when the provider is `kafka`**.
 broker_endpoints = ["127.0.0.1:9092"]
 
+## The connect timeout for kafka client.
+## **It's only used when the provider is `kafka`**.
+#+ connect_timeout = "3s"
+
+## The timeout for kafka client.
+## **It's only used when the provider is `kafka`**.
+#+ timeout = "3s"
+
 ## The max size of a single producer batch.
 ## Warning: Kafka has a default limit of 1MB per message in a topic.
 ## **It's only used when the provider is `kafka`**.
@@ -225,6 +233,7 @@ overwrite_entry_start_id = false
 # endpoint = "https://s3.amazonaws.com"
 # region = "us-west-2"
 # enable_virtual_host_style = false
+# disable_ec2_metadata = false
 
 # Example of using Oss as the storage.
 # [storage]

config/frontend.example.toml

@@ -131,7 +131,6 @@ key_path = ""
 ## For now, gRPC tls config does not support auto reload.
 watch = false
 
-
 ## MySQL server options.
 [mysql]
 ## Whether to enable.

config/metasrv.example.toml

@@ -50,9 +50,6 @@ meta_election_lock_id = 1
 ## For details, please see "https://docs.greptime.com/developer-guide/metasrv/selector".
 selector = "round_robin"
 
-## Store data in memory.
-use_memory_store = false
-
 ## Whether to enable region failover.
 ## This feature is only available on GreptimeDB running on cluster mode and
 ## - Using Remote WAL
@@ -71,6 +68,13 @@ allow_region_failover_on_local_wal = false
 ## Max allowed idle time before removing node info from metasrv memory.
 node_max_idle_time = "24hours"
 
+## Base heartbeat interval for calculating distributed time constants.
+## The frontend heartbeat interval is 6 times of the base heartbeat interval.
+## The flownode/datanode heartbeat interval is 1 times of the base heartbeat interval.
+## e.g., If the base heartbeat interval is 3s, the frontend heartbeat interval is 18s, the flownode/datanode heartbeat interval is 3s.
+## If you change this value, you need to change the heartbeat interval of the flownode/frontend/datanode accordingly.
+#+ heartbeat_interval = "3s"
+
 ## Whether to enable greptimedb telemetry. Enabled by default.
 #+ enable_telemetry = true
 
@@ -109,6 +113,16 @@ key_path = ""
 ## Like "/path/to/ca.crt"
 ca_cert_path = ""
 
+## The backend client options.
+## Currently, only applicable when using etcd as the metadata store.
+#+ [backend_client]
+## The keep alive timeout for backend client.
+#+ keep_alive_timeout = "3s"
+## The keep alive interval for backend client.
+#+ keep_alive_interval = "10s"
+## The connect timeout for backend client.
+#+ connect_timeout = "3s"
+
 ## The gRPC server options.
 [grpc]
 ## The address to bind the gRPC server.
@@ -123,6 +137,10 @@ runtime_size = 8
 max_recv_message_size = "512MB"
 ## The maximum send message size for gRPC server.
 max_send_message_size = "512MB"
+## The server side HTTP/2 keep-alive interval
+#+ http2_keep_alive_interval = "10s"
+## The server side HTTP/2 keep-alive timeout.
+#+ http2_keep_alive_timeout = "3s"
 
 ## The HTTP server options.
 [http]

config/standalone.example.toml

@@ -230,6 +230,14 @@ recovery_parallelism = 2
 ## **It's only used when the provider is `kafka`**.
 broker_endpoints = ["127.0.0.1:9092"]
 
+## The connect timeout for kafka client.
+## **It's only used when the provider is `kafka`**.
+#+ connect_timeout = "3s"
+
+## The timeout for kafka client.
+## **It's only used when the provider is `kafka`**.
+#+ timeout = "3s"
+
 ## Automatically create topics for WAL.
 ## Set to `true` to automatically create topics for WAL.
 ## Otherwise, use topics named `topic_name_prefix_[0..num_topics)`
@@ -332,6 +340,7 @@ max_running_procedures = 128
 # endpoint = "https://s3.amazonaws.com"
 # region = "us-west-2"
 # enable_virtual_host_style = false
+# disable_ec2_metadata = false
 
 # Example of using Oss as the storage.
 # [storage]

src/cli/Cargo.toml

@@ -67,6 +67,7 @@ tracing-appender.workspace = true
 
 [dev-dependencies]
 common-meta = { workspace = true, features = ["testing"] }
+common-test-util.workspace = true
 common-version.workspace = true
 serde.workspace = true
 tempfile.workspace = true

src/cli/src/common.rs

@@ -15,5 +15,8 @@
 mod object_store;
 mod store;
 
-pub use object_store::{ObjectStoreConfig, new_fs_object_store};
+pub use object_store::{
+    ObjectStoreConfig, PrefixedAzblobConnection, PrefixedGcsConnection, PrefixedOssConnection,
+    PrefixedS3Connection, new_fs_object_store,
+};
 pub use store::StoreConfig;

src/cli/src/common/object_store.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use common_base::secrets::SecretString;
+use common_base::secrets::{ExposeSecret, SecretString};
 use common_error::ext::BoxedError;
 use object_store::services::{Azblob, Fs, Gcs, Oss, S3};
 use object_store::util::{with_instrument_layers, with_retry_layers};
@@ -22,9 +22,69 @@ use snafu::ResultExt;
 
 use crate::error::{self};
 
+/// Trait to convert CLI field types to target struct field types.
+/// This enables `Option<SecretString>` (CLI) -> `SecretString` (target) conversions,
+/// allowing us to distinguish "not provided" from "provided but empty".
+trait IntoField<T> {
+    fn into_field(self) -> T;
+}
+
+/// Identity conversion for types that are the same.
+impl<T> IntoField<T> for T {
+    fn into_field(self) -> T {
+        self
+    }
+}
+
+/// Convert `Option<SecretString>` to `SecretString`, using default for None.
+impl IntoField<SecretString> for Option<SecretString> {
+    fn into_field(self) -> SecretString {
+        self.unwrap_or_default()
+    }
+}
+
+/// Trait for checking if a field is effectively empty.
+///
+/// **`is_empty()`**: Checks if the field has no meaningful value
+/// - Used when backend is enabled to validate required fields
+/// - `None`, `Some("")`, `false`, or `""` are considered empty
+trait FieldValidator {
+    /// Check if the field is empty (has no meaningful value).
+    fn is_empty(&self) -> bool;
+}
+
+/// String fields: empty if the string is empty
+impl FieldValidator for String {
+    fn is_empty(&self) -> bool {
+        self.is_empty()
+    }
+}
+
+/// Bool fields: false is considered "empty", true is "provided"
+impl FieldValidator for bool {
+    fn is_empty(&self) -> bool {
+        !self
+    }
+}
+
+/// Option<String> fields: None or empty content is empty
+impl FieldValidator for Option<String> {
+    fn is_empty(&self) -> bool {
+        self.as_ref().is_none_or(|s| s.is_empty())
+    }
+}
+
+/// Option<SecretString> fields: None or empty secret is empty
+/// For secrets, Some("") is treated as "not provided" for both checks
+impl FieldValidator for Option<SecretString> {
+    fn is_empty(&self) -> bool {
+        self.as_ref().is_none_or(|s| s.expose_secret().is_empty())
+    }
+}
+
 macro_rules! wrap_with_clap_prefix {
     (
-        $new_name:ident, $prefix:literal, $base:ty, {
+        $new_name:ident, $prefix:literal, $enable_flag:literal, $base:ty, {
             $( $( #[doc = $doc:expr] )? $( #[alias = $alias:literal] )? $field:ident : $type:ty $( = $default:expr )? ),* $(,)?
         }
     ) => {
@@ -34,15 +94,16 @@ macro_rules! wrap_with_clap_prefix {
                 $(
                     $( #[doc = $doc] )?
                     $( #[clap(alias = $alias)] )?
-                    #[clap(long $(, default_value_t = $default )? )]
-                    [<$prefix $field>]: $type,
+                    #[clap(long, requires = $enable_flag $(, default_value_t = $default )? )]
+                    pub [<$prefix $field>]: $type,
                 )*
             }
 
             impl From<$new_name> for $base {
                 fn from(w: $new_name) -> Self {
                     Self {
-                        $( $field: w.[<$prefix $field>] ),*
+                        // Use into_field() to handle Option<SecretString> -> SecretString conversion
+                        $( $field: w.[<$prefix $field>].into_field() ),*
                     }
                 }
             }
@@ -50,9 +111,90 @@ macro_rules! wrap_with_clap_prefix {
     };
 }
 
+/// Macro for declarative backend validation.
+///
+/// # Validation Rules
+///
+/// For each storage backend (S3, OSS, GCS, Azblob), this function validates:
+/// **When backend is enabled** (e.g., `--s3`): All required fields must be non-empty
+///
+/// Note: When backend is disabled, clap's `requires` attribute ensures no configuration
+/// fields can be provided at parse time.
+///
+/// # Syntax
+///
+/// ```ignore
+/// validate_backend!(
+///     enable: self.enable_s3,
+///     name: "S3",
+///     required: [(field1, "name1"), (field2, "name2"), ...],
+///     custom_validator: |missing| { ... }  // optional
+/// )
+/// ```
+///
+/// # Arguments
+///
+/// - `enable`: Boolean expression indicating if backend is enabled
+/// - `name`: Human-readable backend name for error messages
+/// - `required`: Array of (field_ref, field_name) tuples for required fields
+/// - `custom_validator`: Optional closure for complex validation logic
+///
+/// # Example
+///
+/// ```ignore
+/// validate_backend!(
+///     enable: self.enable_s3,
+///     name: "S3",
+///     required: [
+///         (&self.s3.s3_bucket, "bucket"),
+///         (&self.s3.s3_access_key_id, "access key ID"),
+///     ]
+/// )
+/// ```
+macro_rules! validate_backend {
+    (
+        enable: $enable:expr,
+        name: $backend_name:expr,
+        required: [ $( ($field:expr, $field_name:expr) ),* $(,)? ]
+        $(, custom_validator: $custom_validator:expr)?
+    ) => {{
+        if $enable {
+            // Check required fields when backend is enabled
+            let mut missing = Vec::new();
+            $(
+                if FieldValidator::is_empty($field) {
+                    missing.push($field_name);
+                }
+            )*
+
+            // Run custom validation if provided
+            $(
+                $custom_validator(&mut missing);
+            )?
+
+            if !missing.is_empty() {
+                return Err(BoxedError::new(
+                    error::MissingConfigSnafu {
+                        msg: format!(
+                            "{} {} must be set when --{} is enabled.",
+                            $backend_name,
+                            missing.join(", "),
+                            $backend_name.to_lowercase()
+                        ),
+                    }
+                    .build(),
+                ));
+            }
+        }
+
+        Ok(())
+    }};
+}
+
 wrap_with_clap_prefix! {
     PrefixedAzblobConnection,
     "azblob-",
+    "enable_azblob",
     AzblobConnection,
     {
         #[doc = "The container of the object store."]
@@ -60,9 +202,9 @@ wrap_with_clap_prefix! {
         #[doc = "The root of the object store."]
         root: String = Default::default(),
         #[doc = "The account name of the object store."]
-        account_name: SecretString = Default::default(),
+        account_name: Option<SecretString>,
         #[doc = "The account key of the object store."]
-        account_key: SecretString = Default::default(),
+        account_key: Option<SecretString>,
         #[doc = "The endpoint of the object store."]
         endpoint: String = Default::default(),
         #[doc = "The SAS token of the object store."]
@@ -70,9 +212,33 @@ wrap_with_clap_prefix! {
     }
 }
 
+impl PrefixedAzblobConnection {
+    pub fn validate(&self) -> Result<(), BoxedError> {
+        validate_backend!(
+            enable: true,
+            name: "AzBlob",
+            required: [
+                (&self.azblob_container, "container"),
+                (&self.azblob_root, "root"),
+                (&self.azblob_account_name, "account name"),
+                (&self.azblob_endpoint, "endpoint"),
+            ],
+            custom_validator: |missing: &mut Vec<&str>| {
+                // account_key is only required if sas_token is not provided
+                if self.azblob_sas_token.is_none()
+                    && self.azblob_account_key.is_empty()
+                {
+                    missing.push("account key (when sas_token is not provided)");
+                }
+            }
+        )
+    }
+}
+
 wrap_with_clap_prefix! {
     PrefixedS3Connection,
     "s3-",
+    "enable_s3",
     S3Connection,
     {
         #[doc = "The bucket of the object store."]
@@ -80,21 +246,39 @@ wrap_with_clap_prefix! {
         #[doc = "The root of the object store."]
         root: String = Default::default(),
         #[doc = "The access key ID of the object store."]
-        access_key_id: SecretString = Default::default(),
+        access_key_id: Option<SecretString>,
         #[doc = "The secret access key of the object store."]
-        secret_access_key: SecretString = Default::default(),
+        secret_access_key: Option<SecretString>,
         #[doc = "The endpoint of the object store."]
         endpoint: Option<String>,
         #[doc = "The region of the object store."]
         region: Option<String>,
         #[doc = "Enable virtual host style for the object store."]
         enable_virtual_host_style: bool = Default::default(),
+        #[doc = "Disable EC2 metadata service for the object store."]
+        disable_ec2_metadata: bool = Default::default(),
+    }
+}
+
+impl PrefixedS3Connection {
+    pub fn validate(&self) -> Result<(), BoxedError> {
+        validate_backend!(
+            enable: true,
+            name: "S3",
+            required: [
+                (&self.s3_bucket, "bucket"),
+                (&self.s3_access_key_id, "access key ID"),
+                (&self.s3_secret_access_key, "secret access key"),
+                (&self.s3_region, "region"),
+            ]
+        )
     }
 }
 
 wrap_with_clap_prefix! {
     PrefixedOssConnection,
     "oss-",
+    "enable_oss",
     OssConnection,
     {
         #[doc = "The bucket of the object store."]
@@ -102,17 +286,33 @@ wrap_with_clap_prefix! {
         #[doc = "The root of the object store."]
         root: String = Default::default(),
         #[doc = "The access key ID of the object store."]
-        access_key_id: SecretString = Default::default(),
+        access_key_id: Option<SecretString>,
         #[doc = "The access key secret of the object store."]
-        access_key_secret: SecretString = Default::default(),
+        access_key_secret: Option<SecretString>,
         #[doc = "The endpoint of the object store."]
         endpoint: String = Default::default(),
     }
 }
 
+impl PrefixedOssConnection {
+    pub fn validate(&self) -> Result<(), BoxedError> {
+        validate_backend!(
+            enable: true,
+            name: "OSS",
+            required: [
+                (&self.oss_bucket, "bucket"),
+                (&self.oss_access_key_id, "access key ID"),
+                (&self.oss_access_key_secret, "access key secret"),
+                (&self.oss_endpoint, "endpoint"),
+            ]
+        )
+    }
+}
+
 wrap_with_clap_prefix! {
     PrefixedGcsConnection,
     "gcs-",
+    "enable_gcs",
     GcsConnection,
     {
         #[doc = "The root of the object store."]
@@ -122,40 +322,72 @@ wrap_with_clap_prefix! {
         #[doc = "The scope of the object store."]
         scope: String = Default::default(),
         #[doc = "The credential path of the object store."]
-        credential_path: SecretString = Default::default(),
+        credential_path: Option<SecretString>,
         #[doc = "The credential of the object store."]
-        credential: SecretString = Default::default(),
+        credential: Option<SecretString>,
         #[doc = "The endpoint of the object store."]
         endpoint: String = Default::default(),
     }
 }
 
-/// common config for object store.
+impl PrefixedGcsConnection {
+    pub fn validate(&self) -> Result<(), BoxedError> {
+        validate_backend!(
+            enable: true,
+            name: "GCS",
+            required: [
+                (&self.gcs_bucket, "bucket"),
+                (&self.gcs_root, "root"),
+                (&self.gcs_scope, "scope"),
+            ]
+            // No custom_validator needed: GCS supports Application Default Credentials (ADC)
+            // where neither credential_path nor credential is required.
+            // Endpoint is also optional (defaults to https://storage.googleapis.com).
+        )
+    }
+}
+
+/// Common config for object store.
+///
+/// # Dependency Enforcement
+///
+/// Each backend's configuration fields (e.g., `--s3-bucket`) requires its corresponding
+/// enable flag (e.g., `--s3`) to be present. This is enforced by `clap` at parse time
+/// using the `requires` attribute.
+///
+/// For example, attempting to use `--s3-bucket my-bucket` without `--s3` will result in:
+/// ```text
+/// error: The argument '--s3-bucket <BUCKET>' requires '--s3'
+/// ```
+///
+/// This ensures that users cannot accidentally provide backend-specific configuration
+/// without explicitly enabling that backend.
 #[derive(clap::Parser, Debug, Clone, PartialEq, Default)]
+#[clap(group(clap::ArgGroup::new("storage_backend").required(false).multiple(false)))]
 pub struct ObjectStoreConfig {
     /// Whether to use S3 object store.
-    #[clap(long, alias = "s3")]
+    #[clap(long = "s3", group = "storage_backend")]
     pub enable_s3: bool,
 
     #[clap(flatten)]
     pub s3: PrefixedS3Connection,
 
     /// Whether to use OSS.
-    #[clap(long, alias = "oss")]
+    #[clap(long = "oss", group = "storage_backend")]
     pub enable_oss: bool,
 
     #[clap(flatten)]
     pub oss: PrefixedOssConnection,
 
     /// Whether to use GCS.
-    #[clap(long, alias = "gcs")]
+    #[clap(long = "gcs", group = "storage_backend")]
     pub enable_gcs: bool,
 
     #[clap(flatten)]
     pub gcs: PrefixedGcsConnection,
 
     /// Whether to use Azure Blob.
-    #[clap(long, alias = "azblob")]
+    #[clap(long = "azblob", group = "storage_backend")]
     pub enable_azblob: bool,
 
     #[clap(flatten)]
@@ -173,52 +405,66 @@ pub fn new_fs_object_store(root: &str) -> std::result::Result<ObjectStore, Boxed
     Ok(with_instrument_layers(object_store, false))
 }
 
+macro_rules! gen_object_store_builder {
+    ($method:ident, $field:ident, $conn_type:ty, $service_type:ty) => {
+        pub fn $method(&self) -> Result<ObjectStore, BoxedError> {
+            let config = <$conn_type>::from(self.$field.clone());
+            common_telemetry::info!(
+                "Building object store with {}: {:?}",
+                stringify!($field),
+                config
+            );
+            let object_store = ObjectStore::new(<$service_type>::from(&config))
+                .context(error::InitBackendSnafu)
+                .map_err(BoxedError::new)?
+                .finish();
+            Ok(with_instrument_layers(
+                with_retry_layers(object_store),
+                false,
+            ))
+        }
+    };
+}
+
 impl ObjectStoreConfig {
+    gen_object_store_builder!(build_s3, s3, S3Connection, S3);
+
+    gen_object_store_builder!(build_oss, oss, OssConnection, Oss);
+
+    gen_object_store_builder!(build_gcs, gcs, GcsConnection, Gcs);
+
+    gen_object_store_builder!(build_azblob, azblob, AzblobConnection, Azblob);
+
+    pub fn validate(&self) -> Result<(), BoxedError> {
+        if self.enable_s3 {
+            self.s3.validate()?;
+        }
+        if self.enable_oss {
+            self.oss.validate()?;
+        }
+        if self.enable_gcs {
+            self.gcs.validate()?;
+        }
+        if self.enable_azblob {
+            self.azblob.validate()?;
+        }
+        Ok(())
+    }
+
     /// Builds the object store from the config.
     pub fn build(&self) -> Result<Option<ObjectStore>, BoxedError> {
-        let object_store = if self.enable_s3 {
-            let s3 = S3Connection::from(self.s3.clone());
-            common_telemetry::info!("Building object store with s3: {:?}", s3);
-            Some(
-                ObjectStore::new(S3::from(&s3))
-                    .context(error::InitBackendSnafu)
-                    .map_err(BoxedError::new)?
-                    .finish(),
-            )
+        self.validate()?;
+
+        if self.enable_s3 {
+            self.build_s3().map(Some)
         } else if self.enable_oss {
-            let oss = OssConnection::from(self.oss.clone());
-            common_telemetry::info!("Building object store with oss: {:?}", oss);
-            Some(
-                ObjectStore::new(Oss::from(&oss))
-                    .context(error::InitBackendSnafu)
-                    .map_err(BoxedError::new)?
-                    .finish(),
-            )
+            self.build_oss().map(Some)
         } else if self.enable_gcs {
-            let gcs = GcsConnection::from(self.gcs.clone());
-            common_telemetry::info!("Building object store with gcs: {:?}", gcs);
-            Some(
-                ObjectStore::new(Gcs::from(&gcs))
-                    .context(error::InitBackendSnafu)
-                    .map_err(BoxedError::new)?
-                    .finish(),
-            )
+            self.build_gcs().map(Some)
         } else if self.enable_azblob {
-            let azblob = AzblobConnection::from(self.azblob.clone());
-            common_telemetry::info!("Building object store with azblob: {:?}", azblob);
-            Some(
-                ObjectStore::new(Azblob::from(&azblob))
-                    .context(error::InitBackendSnafu)
-                    .map_err(BoxedError::new)?
-                    .finish(),
-            )
+            self.build_azblob().map(Some)
         } else {
-            None
-        };
-
-        let object_store = object_store
-            .map(|object_store| with_instrument_layers(with_retry_layers(object_store), false));
-
-        Ok(object_store)
+            Ok(None)
+        }
     }
 }

src/cli/src/common/store.rs

@@ -19,7 +19,7 @@ use common_error::ext::BoxedError;
 use common_meta::kv_backend::KvBackendRef;
 use common_meta::kv_backend::chroot::ChrootKvBackend;
 use common_meta::kv_backend::etcd::EtcdStore;
-use meta_srv::metasrv::BackendImpl;
+use meta_srv::metasrv::{BackendClientOptions, BackendImpl};
 use meta_srv::utils::etcd::create_etcd_client_with_tls;
 use servers::tls::{TlsMode, TlsOption};
 
@@ -112,9 +112,13 @@ impl StoreConfig {
             let kvbackend = match self.backend {
                 BackendImpl::EtcdStore => {
                     let tls_config = self.tls_config();
-                    let etcd_client = create_etcd_client_with_tls(store_addrs, tls_config.as_ref())
-                        .await
-                        .map_err(BoxedError::new)?;
+                    let etcd_client = create_etcd_client_with_tls(
+                        store_addrs,
+                        &BackendClientOptions::default(),
+                        tls_config.as_ref(),
+                    )
+                    .await
+                    .map_err(BoxedError::new)?;
                     Ok(EtcdStore::with_etcd_client(etcd_client, max_txn_ops))
                 }
                 #[cfg(feature = "pg_kvbackend")]

src/cli/src/data.rs

@@ -14,6 +14,7 @@
 
 mod export;
 mod import;
+mod storage_export;
 
 use clap::Subcommand;
 use client::DEFAULT_CATALOG_NAME;

src/cli/src/data/storage_export.rs

@@ -0,0 +1,373 @@
+// 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::path::PathBuf;
+
+use common_base::secrets::{ExposeSecret, SecretString};
+use common_error::ext::BoxedError;
+
+use crate::common::{
+    PrefixedAzblobConnection, PrefixedGcsConnection, PrefixedOssConnection, PrefixedS3Connection,
+};
+
+/// Helper function to extract secret string from Option<SecretString>.
+/// Returns empty string if None.
+fn expose_optional_secret(secret: &Option<SecretString>) -> &str {
+    secret
+        .as_ref()
+        .map(|s| s.expose_secret().as_str())
+        .unwrap_or("")
+}
+
+/// Helper function to format root path with leading slash if non-empty.
+fn format_root_path(root: &str) -> String {
+    if root.is_empty() {
+        String::new()
+    } else {
+        format!("/{}", root)
+    }
+}
+
+/// Helper function to mask multiple secrets in a string.
+fn mask_secrets(mut sql: String, secrets: &[&str]) -> String {
+    for secret in secrets {
+        if !secret.is_empty() {
+            sql = sql.replace(secret, "[REDACTED]");
+        }
+    }
+    sql
+}
+
+/// Helper function to format storage URI.
+fn format_uri(scheme: &str, bucket: &str, root: &str, path: &str) -> String {
+    let root = format_root_path(root);
+    format!("{}://{}{}/{}", scheme, bucket, root, path)
+}
+
+/// Trait for storage backends that can be used for data export.
+pub trait StorageExport: Send + Sync {
+    /// Generate the storage path for COPY DATABASE command.
+    /// Returns (path, connection_string) where connection_string includes CONNECTION clause.
+    fn get_storage_path(&self, catalog: &str, schema: &str) -> (String, String);
+
+    /// Format the output path for logging purposes.
+    fn format_output_path(&self, file_path: &str) -> String;
+
+    /// Mask sensitive information in SQL commands for safe logging.
+    fn mask_sensitive_info(&self, sql: &str) -> String;
+}
+
+macro_rules! define_backend {
+    ($name:ident, $config:ty) => {
+        #[derive(Clone)]
+        pub struct $name {
+            config: $config,
+        }
+
+        impl $name {
+            pub fn new(config: $config) -> Result<Self, BoxedError> {
+                config.validate()?;
+                Ok(Self { config })
+            }
+        }
+    };
+}
+
+/// Local file system storage backend.
+#[derive(Clone)]
+pub struct FsBackend {
+    output_dir: String,
+}
+
+impl FsBackend {
+    pub fn new(output_dir: String) -> Self {
+        Self { output_dir }
+    }
+}
+
+impl StorageExport for FsBackend {
+    fn get_storage_path(&self, catalog: &str, schema: &str) -> (String, String) {
+        if self.output_dir.is_empty() {
+            unreachable!("output_dir must be set when not using remote storage")
+        }
+        let path = PathBuf::from(&self.output_dir)
+            .join(catalog)
+            .join(format!("{schema}/"))
+            .to_string_lossy()
+            .to_string();
+        (path, String::new())
+    }
+
+    fn format_output_path(&self, file_path: &str) -> String {
+        format!("{}/{}", self.output_dir, file_path)
+    }
+
+    fn mask_sensitive_info(&self, sql: &str) -> String {
+        sql.to_string()
+    }
+}
+
+define_backend!(S3Backend, PrefixedS3Connection);
+
+impl StorageExport for S3Backend {
+    fn get_storage_path(&self, catalog: &str, schema: &str) -> (String, String) {
+        let s3_path = format_uri(
+            "s3",
+            &self.config.s3_bucket,
+            &self.config.s3_root,
+            &format!("{}/{}/", catalog, schema),
+        );
+
+        let mut connection_options = vec![
+            format!(
+                "ACCESS_KEY_ID='{}'",
+                expose_optional_secret(&self.config.s3_access_key_id)
+            ),
+            format!(
+                "SECRET_ACCESS_KEY='{}'",
+                expose_optional_secret(&self.config.s3_secret_access_key)
+            ),
+        ];
+
+        if let Some(region) = &self.config.s3_region {
+            connection_options.push(format!("REGION='{}'", region));
+        }
+
+        if let Some(endpoint) = &self.config.s3_endpoint {
+            connection_options.push(format!("ENDPOINT='{}'", endpoint));
+        }
+
+        let connection_str = format!(" CONNECTION ({})", connection_options.join(", "));
+        (s3_path, connection_str)
+    }
+
+    fn format_output_path(&self, file_path: &str) -> String {
+        format_uri(
+            "s3",
+            &self.config.s3_bucket,
+            &self.config.s3_root,
+            file_path,
+        )
+    }
+
+    fn mask_sensitive_info(&self, sql: &str) -> String {
+        mask_secrets(
+            sql.to_string(),
+            &[
+                expose_optional_secret(&self.config.s3_access_key_id),
+                expose_optional_secret(&self.config.s3_secret_access_key),
+            ],
+        )
+    }
+}
+
+define_backend!(OssBackend, PrefixedOssConnection);
+
+impl StorageExport for OssBackend {
+    fn get_storage_path(&self, catalog: &str, schema: &str) -> (String, String) {
+        let oss_path = format_uri(
+            "oss",
+            &self.config.oss_bucket,
+            &self.config.oss_root,
+            &format!("{}/{}/", catalog, schema),
+        );
+
+        let connection_options = [
+            format!(
+                "ACCESS_KEY_ID='{}'",
+                expose_optional_secret(&self.config.oss_access_key_id)
+            ),
+            format!(
+                "ACCESS_KEY_SECRET='{}'",
+                expose_optional_secret(&self.config.oss_access_key_secret)
+            ),
+        ];
+
+        let connection_str = format!(" CONNECTION ({})", connection_options.join(", "));
+        (oss_path, connection_str)
+    }
+
+    fn format_output_path(&self, file_path: &str) -> String {
+        format_uri(
+            "oss",
+            &self.config.oss_bucket,
+            &self.config.oss_root,
+            file_path,
+        )
+    }
+
+    fn mask_sensitive_info(&self, sql: &str) -> String {
+        mask_secrets(
+            sql.to_string(),
+            &[
+                expose_optional_secret(&self.config.oss_access_key_id),
+                expose_optional_secret(&self.config.oss_access_key_secret),
+            ],
+        )
+    }
+}
+
+define_backend!(GcsBackend, PrefixedGcsConnection);
+
+impl StorageExport for GcsBackend {
+    fn get_storage_path(&self, catalog: &str, schema: &str) -> (String, String) {
+        let gcs_path = format_uri(
+            "gcs",
+            &self.config.gcs_bucket,
+            &self.config.gcs_root,
+            &format!("{}/{}/", catalog, schema),
+        );
+
+        let mut connection_options = Vec::new();
+
+        let credential_path = expose_optional_secret(&self.config.gcs_credential_path);
+        if !credential_path.is_empty() {
+            connection_options.push(format!("CREDENTIAL_PATH='{}'", credential_path));
+        }
+
+        let credential = expose_optional_secret(&self.config.gcs_credential);
+        if !credential.is_empty() {
+            connection_options.push(format!("CREDENTIAL='{}'", credential));
+        }
+
+        if !self.config.gcs_endpoint.is_empty() {
+            connection_options.push(format!("ENDPOINT='{}'", self.config.gcs_endpoint));
+        }
+
+        let connection_str = if connection_options.is_empty() {
+            String::new()
+        } else {
+            format!(" CONNECTION ({})", connection_options.join(", "))
+        };
+
+        (gcs_path, connection_str)
+    }
+
+    fn format_output_path(&self, file_path: &str) -> String {
+        format_uri(
+            "gcs",
+            &self.config.gcs_bucket,
+            &self.config.gcs_root,
+            file_path,
+        )
+    }
+
+    fn mask_sensitive_info(&self, sql: &str) -> String {
+        mask_secrets(
+            sql.to_string(),
+            &[
+                expose_optional_secret(&self.config.gcs_credential_path),
+                expose_optional_secret(&self.config.gcs_credential),
+            ],
+        )
+    }
+}
+
+define_backend!(AzblobBackend, PrefixedAzblobConnection);
+
+impl StorageExport for AzblobBackend {
+    fn get_storage_path(&self, catalog: &str, schema: &str) -> (String, String) {
+        let azblob_path = format_uri(
+            "azblob",
+            &self.config.azblob_container,
+            &self.config.azblob_root,
+            &format!("{}/{}/", catalog, schema),
+        );
+
+        let mut connection_options = vec![
+            format!(
+                "ACCOUNT_NAME='{}'",
+                expose_optional_secret(&self.config.azblob_account_name)
+            ),
+            format!(
+                "ACCOUNT_KEY='{}'",
+                expose_optional_secret(&self.config.azblob_account_key)
+            ),
+        ];
+
+        if let Some(sas_token) = &self.config.azblob_sas_token {
+            connection_options.push(format!("SAS_TOKEN='{}'", sas_token));
+        }
+
+        let connection_str = format!(" CONNECTION ({})", connection_options.join(", "));
+        (azblob_path, connection_str)
+    }
+
+    fn format_output_path(&self, file_path: &str) -> String {
+        format_uri(
+            "azblob",
+            &self.config.azblob_container,
+            &self.config.azblob_root,
+            file_path,
+        )
+    }
+
+    fn mask_sensitive_info(&self, sql: &str) -> String {
+        mask_secrets(
+            sql.to_string(),
+            &[
+                expose_optional_secret(&self.config.azblob_account_name),
+                expose_optional_secret(&self.config.azblob_account_key),
+            ],
+        )
+    }
+}
+
+#[derive(Clone)]
+pub enum StorageType {
+    Fs(FsBackend),
+    S3(S3Backend),
+    Oss(OssBackend),
+    Gcs(GcsBackend),
+    Azblob(AzblobBackend),
+}
+
+impl StorageExport for StorageType {
+    fn get_storage_path(&self, catalog: &str, schema: &str) -> (String, String) {
+        match self {
+            StorageType::Fs(backend) => backend.get_storage_path(catalog, schema),
+            StorageType::S3(backend) => backend.get_storage_path(catalog, schema),
+            StorageType::Oss(backend) => backend.get_storage_path(catalog, schema),
+            StorageType::Gcs(backend) => backend.get_storage_path(catalog, schema),
+            StorageType::Azblob(backend) => backend.get_storage_path(catalog, schema),
+        }
+    }
+
+    fn format_output_path(&self, file_path: &str) -> String {
+        match self {
+            StorageType::Fs(backend) => backend.format_output_path(file_path),
+            StorageType::S3(backend) => backend.format_output_path(file_path),
+            StorageType::Oss(backend) => backend.format_output_path(file_path),
+            StorageType::Gcs(backend) => backend.format_output_path(file_path),
+            StorageType::Azblob(backend) => backend.format_output_path(file_path),
+        }
+    }
+
+    fn mask_sensitive_info(&self, sql: &str) -> String {
+        match self {
+            StorageType::Fs(backend) => backend.mask_sensitive_info(sql),
+            StorageType::S3(backend) => backend.mask_sensitive_info(sql),
+            StorageType::Oss(backend) => backend.mask_sensitive_info(sql),
+            StorageType::Gcs(backend) => backend.mask_sensitive_info(sql),
+            StorageType::Azblob(backend) => backend.mask_sensitive_info(sql),
+        }
+    }
+}
+
+impl StorageType {
+    /// Returns true if the storage backend is remote (not local filesystem).
+    pub fn is_remote_storage(&self) -> bool {
+        !matches!(self, StorageType::Fs(_))
+    }
+}

src/cli/src/error.rs

@@ -253,12 +253,6 @@ pub enum Error {
         error: ObjectStoreError,
     },
 
-    #[snafu(display("S3 config need be set"))]
-    S3ConfigNotSet {
-        #[snafu(implicit)]
-        location: Location,
-    },
-
     #[snafu(display("Output directory not set"))]
     OutputDirNotSet {
         #[snafu(implicit)]
@@ -364,9 +358,9 @@ impl ErrorExt for Error {
 
             Error::Other { source, .. } => source.status_code(),
             Error::OpenDal { .. } | Error::InitBackend { .. } => StatusCode::Internal,
-            Error::S3ConfigNotSet { .. }
-            | Error::OutputDirNotSet { .. }
-            | Error::EmptyStoreAddrs { .. } => StatusCode::InvalidArguments,
+            Error::OutputDirNotSet { .. } | Error::EmptyStoreAddrs { .. } => {
+                StatusCode::InvalidArguments
+            }
 
             Error::BuildRuntime { source, .. } => source.status_code(),
 

src/cmd/src/metasrv.rs

@@ -20,6 +20,7 @@ use async_trait::async_trait;
 use clap::Parser;
 use common_base::Plugins;
 use common_config::Configurable;
+use common_meta::distributed_time_constants::init_distributed_time_constants;
 use common_telemetry::info;
 use common_telemetry::logging::{DEFAULT_LOGGING_DIR, TracingOptions};
 use common_version::{short_version, verbose_version};
@@ -154,8 +155,6 @@ pub struct StartCommand {
     #[clap(short, long)]
     selector: Option<String>,
     #[clap(long)]
-    use_memory_store: Option<bool>,
-    #[clap(long)]
     enable_region_failover: Option<bool>,
     #[clap(long)]
     http_addr: Option<String>,
@@ -185,7 +184,6 @@ impl Debug for StartCommand {
             .field("store_addrs", &self.sanitize_store_addrs())
             .field("config_file", &self.config_file)
             .field("selector", &self.selector)
-            .field("use_memory_store", &self.use_memory_store)
             .field("enable_region_failover", &self.enable_region_failover)
             .field("http_addr", &self.http_addr)
             .field("http_timeout", &self.http_timeout)
@@ -267,10 +265,6 @@ impl StartCommand {
                 .context(error::UnsupportedSelectorTypeSnafu { selector_type })?;
         }
 
-        if let Some(use_memory_store) = self.use_memory_store {
-            opts.use_memory_store = use_memory_store;
-        }
-
         if let Some(enable_region_failover) = self.enable_region_failover {
             opts.enable_region_failover = enable_region_failover;
         }
@@ -327,6 +321,7 @@ impl StartCommand {
         log_versions(verbose_version(), short_version(), APP_NAME);
         maybe_activate_heap_profile(&opts.component.memory);
         create_resource_limit_metrics(APP_NAME);
+        init_distributed_time_constants(opts.component.heartbeat_interval);
 
         info!("Metasrv start command: {:#?}", self);
 
@@ -389,7 +384,6 @@ mod tests {
             server_addr = "127.0.0.1:3002"
             store_addr = "127.0.0.1:2379"
             selector = "LeaseBased"
-            use_memory_store = false
 
             [logging]
             level = "debug"
@@ -468,7 +462,6 @@ mod tests {
             server_addr = "127.0.0.1:3002"
             datanode_lease_secs = 15
             selector = "LeaseBased"
-            use_memory_store = false
 
             [http]
             addr = "127.0.0.1:4000"

src/common/config/src/error.rs

@@ -59,15 +59,6 @@ pub enum Error {
         location: Location,
     },
 
-    #[snafu(display("Failed to canonicalize path: {}", path))]
-    CanonicalizePath {
-        path: String,
-        #[snafu(source)]
-        error: std::io::Error,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
     #[snafu(display("Invalid path '{}': expected a file, not a directory", path))]
     InvalidPath {
         path: String,
@@ -82,8 +73,7 @@ impl ErrorExt for Error {
             Error::TomlFormat { .. }
             | Error::LoadLayeredConfig { .. }
             | Error::FileWatch { .. }
-            | Error::InvalidPath { .. }
-            | Error::CanonicalizePath { .. } => StatusCode::InvalidArguments,
+            | Error::InvalidPath { .. } => StatusCode::InvalidArguments,
             Error::SerdeJson { .. } => StatusCode::Unexpected,
         }
     }

src/common/config/src/file_watcher.rs

@@ -30,7 +30,7 @@ use common_telemetry::{error, info, warn};
 use notify::{EventKind, RecursiveMode, Watcher};
 use snafu::ResultExt;
 
-use crate::error::{CanonicalizePathSnafu, FileWatchSnafu, InvalidPathSnafu, Result};
+use crate::error::{FileWatchSnafu, InvalidPathSnafu, Result};
 
 /// Configuration for the file watcher behavior.
 #[derive(Debug, Clone, Default)]
@@ -41,15 +41,10 @@ pub struct FileWatcherConfig {
 
 impl FileWatcherConfig {
     pub fn new() -> Self {
-        Self::default()
+        Default::default()
     }
 
-    pub fn with_modify_and_create(mut self) -> Self {
-        self.include_remove_events = false;
-        self
-    }
-
-    pub fn with_remove_events(mut self) -> Self {
+    pub fn include_remove_events(mut self) -> Self {
         self.include_remove_events = true;
         self
     }
@@ -93,11 +88,8 @@ impl FileWatcherBuilder {
                 path: path.display().to_string(),
             }
         );
-        // Canonicalize the path for reliable comparison with event paths
-        let canonical = path.canonicalize().context(CanonicalizePathSnafu {
-            path: path.display().to_string(),
-        })?;
-        self.file_paths.push(canonical);
+
+        self.file_paths.push(path.to_path_buf());
         Ok(self)
     }
 
@@ -144,7 +136,6 @@ impl FileWatcherBuilder {
         }
 
         let config = self.config;
-        let watched_files: HashSet<PathBuf> = self.file_paths.iter().cloned().collect();
 
         info!(
             "Spawning file watcher for paths: {:?} (watching parent directories)",
@@ -165,25 +156,7 @@ impl FileWatcherBuilder {
                             continue;
                         }
 
-                        // Check if any of the event paths match our watched files
-                        let is_watched_file = event.paths.iter().any(|event_path| {
-                            // Try to canonicalize the event path for comparison
-                            // If the file was deleted, canonicalize will fail, so we also
-                            // compare the raw path
-                            if let Ok(canonical) = event_path.canonicalize()
-                                && watched_files.contains(&canonical)
-                            {
-                                return true;
-                            }
-                            // For deleted files, compare using the raw path
-                            watched_files.contains(event_path)
-                        });
-
-                        if !is_watched_file {
-                            continue;
-                        }
-
-                        info!(?event.kind, ?event.paths, "Detected file change");
+                        info!(?event.kind, ?event.paths, "Detected folder change");
                         callback();
                     }
                     Err(err) => {
@@ -301,55 +274,4 @@ mod tests {
             "Watcher should have detected file recreation"
         );
     }
-
-    #[test]
-    fn test_file_watcher_ignores_other_files() {
-        common_telemetry::init_default_ut_logging();
-
-        let dir = create_temp_dir("test_file_watcher_other");
-        let watched_file = dir.path().join("watched.txt");
-        let other_file = dir.path().join("other.txt");
-
-        // Create both files
-        std::fs::write(&watched_file, "watched content").unwrap();
-        std::fs::write(&other_file, "other content").unwrap();
-
-        let counter = Arc::new(AtomicUsize::new(0));
-        let counter_clone = counter.clone();
-
-        FileWatcherBuilder::new()
-            .watch_path(&watched_file)
-            .unwrap()
-            .config(FileWatcherConfig::new())
-            .spawn(move || {
-                counter_clone.fetch_add(1, Ordering::SeqCst);
-            })
-            .unwrap();
-
-        // Give watcher time to start
-        std::thread::sleep(Duration::from_millis(100));
-
-        // Modify the other file - should NOT trigger callback
-        std::fs::write(&other_file, "modified other content").unwrap();
-
-        // Wait for potential event
-        std::thread::sleep(Duration::from_millis(500));
-
-        assert_eq!(
-            counter.load(Ordering::SeqCst),
-            0,
-            "Watcher should not have detected changes to other files"
-        );
-
-        // Now modify the watched file - SHOULD trigger callback
-        std::fs::write(&watched_file, "modified watched content").unwrap();
-
-        // Wait for the event to be processed
-        std::thread::sleep(Duration::from_millis(500));
-
-        assert!(
-            counter.load(Ordering::SeqCst) >= 1,
-            "Watcher should have detected change to watched file"
-        );
-    }
 }

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

@@ -27,6 +27,7 @@ const SECRET_ACCESS_KEY: &str = "secret_access_key";
 const SESSION_TOKEN: &str = "session_token";
 const REGION: &str = "region";
 const ENABLE_VIRTUAL_HOST_STYLE: &str = "enable_virtual_host_style";
+const DISABLE_EC2_METADATA: &str = "disable_ec2_metadata";
 
 pub fn is_supported_in_s3(key: &str) -> bool {
     [
@@ -36,6 +37,7 @@ pub fn is_supported_in_s3(key: &str) -> bool {
         SESSION_TOKEN,
         REGION,
         ENABLE_VIRTUAL_HOST_STYLE,
+        DISABLE_EC2_METADATA,
     ]
     .contains(&key)
 }
@@ -82,6 +84,21 @@ pub fn build_s3_backend(
         }
     }
 
+    if let Some(disable_str) = connection.get(DISABLE_EC2_METADATA) {
+        let disable = disable_str.as_str().parse::<bool>().map_err(|e| {
+            error::InvalidConnectionSnafu {
+                msg: format!(
+                    "failed to parse the option {}={}, {}",
+                    DISABLE_EC2_METADATA, disable_str, e
+                ),
+            }
+            .build()
+        })?;
+        if disable {
+            builder = builder.disable_ec2_metadata();
+        }
+    }
+
     // TODO(weny): Consider finding a better way to eliminate duplicate code.
     Ok(ObjectStore::new(builder)
         .context(error::BuildBackendSnafu)?
@@ -109,6 +126,7 @@ mod tests {
         assert!(is_supported_in_s3(SESSION_TOKEN));
         assert!(is_supported_in_s3(REGION));
         assert!(is_supported_in_s3(ENABLE_VIRTUAL_HOST_STYLE));
+        assert!(is_supported_in_s3(DISABLE_EC2_METADATA));
         assert!(!is_supported_in_s3("foo"))
     }
 }

src/common/meta/src/distributed_time_constants.rs

@@ -12,27 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::OnceLock;
 use std::time::Duration;
 
-use etcd_client::ConnectOptions;
-
-/// Heartbeat interval time (is the basic unit of various time).
-pub const HEARTBEAT_INTERVAL_MILLIS: u64 = 3000;
-
-/// The frontend will also send heartbeats to Metasrv, sending an empty
-/// heartbeat every HEARTBEAT_INTERVAL_MILLIS * 6 seconds.
-pub const FRONTEND_HEARTBEAT_INTERVAL_MILLIS: u64 = HEARTBEAT_INTERVAL_MILLIS * 6;
-
-/// The lease seconds of a region. It's set by 3 heartbeat intervals
-/// (HEARTBEAT_INTERVAL_MILLIS × 3), plus some extra buffer (1 second).
-pub const REGION_LEASE_SECS: u64 =
-    Duration::from_millis(HEARTBEAT_INTERVAL_MILLIS * 3).as_secs() + 1;
-
-/// When creating table or region failover, a target node needs to be selected.
-/// If the node's lease has expired, the `Selector` will not select it.
-pub const DATANODE_LEASE_SECS: u64 = REGION_LEASE_SECS;
-
-pub const FLOWNODE_LEASE_SECS: u64 = DATANODE_LEASE_SECS;
+pub const BASE_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(3);
 
 /// The lease seconds of metasrv leader.
 pub const META_LEASE_SECS: u64 = 5;
@@ -52,14 +35,6 @@ pub const HEARTBEAT_CHANNEL_KEEP_ALIVE_INTERVAL_SECS: Duration = Duration::from_
 /// The keep-alive timeout of the heartbeat channel.
 pub const HEARTBEAT_CHANNEL_KEEP_ALIVE_TIMEOUT_SECS: Duration = Duration::from_secs(5);
 
-/// The default options for the etcd client.
-pub fn default_etcd_client_options() -> ConnectOptions {
-    ConnectOptions::new()
-        .with_keep_alive_while_idle(true)
-        .with_keep_alive(Duration::from_secs(15), Duration::from_secs(5))
-        .with_connect_timeout(Duration::from_secs(10))
-}
-
 /// The default mailbox round-trip timeout.
 pub const MAILBOX_RTT_SECS: u64 = 1;
 
@@ -68,3 +43,60 @@ pub const TOPIC_STATS_REPORT_INTERVAL_SECS: u64 = 15;
 
 /// The retention seconds of topic stats.
 pub const TOPIC_STATS_RETENTION_SECS: u64 = TOPIC_STATS_REPORT_INTERVAL_SECS * 100;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+/// The distributed time constants.
+pub struct DistributedTimeConstants {
+    pub heartbeat_interval: Duration,
+    pub frontend_heartbeat_interval: Duration,
+    pub region_lease: Duration,
+    pub datanode_lease: Duration,
+    pub flownode_lease: Duration,
+}
+
+/// The frontend heartbeat interval is 6 times of the base heartbeat interval.
+pub fn frontend_heartbeat_interval(base_heartbeat_interval: Duration) -> Duration {
+    base_heartbeat_interval * 6
+}
+
+impl DistributedTimeConstants {
+    /// Create a new DistributedTimeConstants from the heartbeat interval.
+    pub fn from_heartbeat_interval(heartbeat_interval: Duration) -> Self {
+        let region_lease = heartbeat_interval * 3 + Duration::from_secs(1);
+        let datanode_lease = region_lease;
+        let flownode_lease = datanode_lease;
+        Self {
+            heartbeat_interval,
+            frontend_heartbeat_interval: frontend_heartbeat_interval(heartbeat_interval),
+            region_lease,
+            datanode_lease,
+            flownode_lease,
+        }
+    }
+}
+
+impl Default for DistributedTimeConstants {
+    fn default() -> Self {
+        Self::from_heartbeat_interval(BASE_HEARTBEAT_INTERVAL)
+    }
+}
+
+static DEFAULT_DISTRIBUTED_TIME_CONSTANTS: OnceLock<DistributedTimeConstants> = OnceLock::new();
+
+/// Get the default distributed time constants.
+pub fn default_distributed_time_constants() -> &'static DistributedTimeConstants {
+    DEFAULT_DISTRIBUTED_TIME_CONSTANTS.get_or_init(Default::default)
+}
+
+/// Initialize the default distributed time constants.
+pub fn init_distributed_time_constants(base_heartbeat_interval: Duration) {
+    let distributed_time_constants =
+        DistributedTimeConstants::from_heartbeat_interval(base_heartbeat_interval);
+    DEFAULT_DISTRIBUTED_TIME_CONSTANTS
+        .set(distributed_time_constants)
+        .expect("Failed to set default distributed time constants");
+    common_telemetry::info!(
+        "Initialized default distributed time constants: {:#?}",
+        distributed_time_constants
+    );
+}

src/common/meta/src/wal_options_allocator/topic_creator.rs

@@ -14,7 +14,7 @@
 
 use common_telemetry::{debug, error, info};
 use common_wal::config::kafka::common::{
-    DEFAULT_BACKOFF_CONFIG, DEFAULT_CONNECT_TIMEOUT, KafkaConnectionConfig, KafkaTopicConfig,
+    DEFAULT_BACKOFF_CONFIG, KafkaConnectionConfig, KafkaTopicConfig,
 };
 use rskafka::client::error::Error as RsKafkaError;
 use rskafka::client::error::ProtocolError::TopicAlreadyExists;
@@ -211,7 +211,8 @@ pub async fn build_kafka_client(connection: &KafkaConnectionConfig) -> Result<Cl
     // Builds an kafka controller client for creating topics.
     let mut builder = ClientBuilder::new(connection.broker_endpoints.clone())
         .backoff_config(DEFAULT_BACKOFF_CONFIG)
-        .connect_timeout(Some(DEFAULT_CONNECT_TIMEOUT));
+        .connect_timeout(Some(connection.connect_timeout))
+        .timeout(Some(connection.timeout));
     if let Some(sasl) = &connection.sasl {
         builder = builder.sasl_config(sasl.config.clone().into_sasl_config());
     };

src/common/telemetry/src/metric.rs

@@ -71,6 +71,7 @@ pub fn convert_metric_to_write_request(
                         timestamp,
                     }],
                     exemplars: vec![],
+                    histograms: vec![],
                 }),
                 MetricType::GAUGE => timeseries.push(TimeSeries {
                     labels: convert_label(m.get_label(), mf_name, None),
@@ -79,6 +80,7 @@ pub fn convert_metric_to_write_request(
                         timestamp,
                     }],
                     exemplars: vec![],
+                    histograms: vec![],
                 }),
                 MetricType::HISTOGRAM => {
                     let h = m.get_histogram();
@@ -97,6 +99,7 @@ pub fn convert_metric_to_write_request(
                                 timestamp,
                             }],
                             exemplars: vec![],
+                            histograms: vec![],
                         });
                         if upper_bound.is_sign_positive() && upper_bound.is_infinite() {
                             inf_seen = true;
@@ -114,6 +117,7 @@ pub fn convert_metric_to_write_request(
                                 timestamp,
                             }],
                             exemplars: vec![],
+                            histograms: vec![],
                         });
                     }
                     timeseries.push(TimeSeries {
@@ -127,6 +131,7 @@ pub fn convert_metric_to_write_request(
                             timestamp,
                         }],
                         exemplars: vec![],
+                        histograms: vec![],
                     });
                     timeseries.push(TimeSeries {
                         labels: convert_label(
@@ -139,6 +144,7 @@ pub fn convert_metric_to_write_request(
                             timestamp,
                         }],
                         exemplars: vec![],
+                        histograms: vec![],
                     });
                 }
                 MetricType::SUMMARY => {
@@ -155,6 +161,7 @@ pub fn convert_metric_to_write_request(
                                 timestamp,
                             }],
                             exemplars: vec![],
+                            histograms: vec![],
                         });
                     }
                     timeseries.push(TimeSeries {
@@ -168,6 +175,7 @@ pub fn convert_metric_to_write_request(
                             timestamp,
                         }],
                         exemplars: vec![],
+                        histograms: vec![],
                     });
                     timeseries.push(TimeSeries {
                         labels: convert_label(
@@ -180,6 +188,7 @@ pub fn convert_metric_to_write_request(
                             timestamp,
                         }],
                         exemplars: vec![],
+                        histograms: vec![],
                     });
                 }
                 MetricType::UNTYPED => {
@@ -274,7 +283,7 @@ mod test {
 
         assert_eq!(
             format!("{:?}", write_quest.timeseries),
-            r#"[TimeSeries { labels: [Label { name: "__name__", value: "test_counter" }, Label { name: "a", value: "1" }, Label { name: "b", value: "2" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [] }]"#
+            r#"[TimeSeries { labels: [Label { name: "__name__", value: "test_counter" }, Label { name: "a", value: "1" }, Label { name: "b", value: "2" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [], histograms: [] }]"#
         );
 
         let gauge_opts = Opts::new("test_gauge", "test help")
@@ -288,7 +297,7 @@ mod test {
         let write_quest = convert_metric_to_write_request(mf, None, 0);
         assert_eq!(
             format!("{:?}", write_quest.timeseries),
-            r#"[TimeSeries { labels: [Label { name: "__name__", value: "test_gauge" }, Label { name: "a", value: "1" }, Label { name: "b", value: "2" }], samples: [Sample { value: 42.0, timestamp: 0 }], exemplars: [] }]"#
+            r#"[TimeSeries { labels: [Label { name: "__name__", value: "test_gauge" }, Label { name: "a", value: "1" }, Label { name: "b", value: "2" }], samples: [Sample { value: 42.0, timestamp: 0 }], exemplars: [], histograms: [] }]"#
         );
     }
 
@@ -305,20 +314,20 @@ mod test {
             .iter()
             .map(|x| format!("{:?}", x))
             .collect();
-        let ans = r#"TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "0.005" }], samples: [Sample { value: 0.0, timestamp: 0 }], exemplars: [] }
-TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "0.01" }], samples: [Sample { value: 0.0, timestamp: 0 }], exemplars: [] }
-TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "0.025" }], samples: [Sample { value: 0.0, timestamp: 0 }], exemplars: [] }
-TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "0.05" }], samples: [Sample { value: 0.0, timestamp: 0 }], exemplars: [] }
-TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "0.1" }], samples: [Sample { value: 0.0, timestamp: 0 }], exemplars: [] }
-TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "0.25" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [] }
-TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "0.5" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [] }
-TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "1" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [] }
-TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "2.5" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [] }
-TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "5" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [] }
-TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "10" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [] }
-TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "+Inf" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [] }
-TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_sum" }, Label { name: "a", value: "1" }], samples: [Sample { value: 0.25, timestamp: 0 }], exemplars: [] }
-TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_count" }, Label { name: "a", value: "1" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [] }"#;
+        let ans = r#"TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "0.005" }], samples: [Sample { value: 0.0, timestamp: 0 }], exemplars: [], histograms: [] }
+TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "0.01" }], samples: [Sample { value: 0.0, timestamp: 0 }], exemplars: [], histograms: [] }
+TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "0.025" }], samples: [Sample { value: 0.0, timestamp: 0 }], exemplars: [], histograms: [] }
+TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "0.05" }], samples: [Sample { value: 0.0, timestamp: 0 }], exemplars: [], histograms: [] }
+TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "0.1" }], samples: [Sample { value: 0.0, timestamp: 0 }], exemplars: [], histograms: [] }
+TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "0.25" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [], histograms: [] }
+TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "0.5" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [], histograms: [] }
+TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "1" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [], histograms: [] }
+TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "2.5" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [], histograms: [] }
+TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "5" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [], histograms: [] }
+TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "10" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [], histograms: [] }
+TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_bucket" }, Label { name: "a", value: "1" }, Label { name: "le", value: "+Inf" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [], histograms: [] }
+TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_sum" }, Label { name: "a", value: "1" }], samples: [Sample { value: 0.25, timestamp: 0 }], exemplars: [], histograms: [] }
+TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_count" }, Label { name: "a", value: "1" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [], histograms: [] }"#;
         assert_eq!(write_quest_str.join("\n"), ans);
     }
 
@@ -355,10 +364,10 @@ TimeSeries { labels: [Label { name: "__name__", value: "test_histogram_count" },
             .iter()
             .map(|x| format!("{:?}", x))
             .collect();
-        let ans = r#"TimeSeries { labels: [Label { name: "__name__", value: "test_summary" }, Label { name: "quantile", value: "50" }], samples: [Sample { value: 3.0, timestamp: 20 }], exemplars: [] }
-TimeSeries { labels: [Label { name: "__name__", value: "test_summary" }, Label { name: "quantile", value: "100" }], samples: [Sample { value: 5.0, timestamp: 20 }], exemplars: [] }
-TimeSeries { labels: [Label { name: "__name__", value: "test_summary_sum" }], samples: [Sample { value: 15.0, timestamp: 20 }], exemplars: [] }
-TimeSeries { labels: [Label { name: "__name__", value: "test_summary_count" }], samples: [Sample { value: 5.0, timestamp: 20 }], exemplars: [] }"#;
+        let ans = r#"TimeSeries { labels: [Label { name: "__name__", value: "test_summary" }, Label { name: "quantile", value: "50" }], samples: [Sample { value: 3.0, timestamp: 20 }], exemplars: [], histograms: [] }
+TimeSeries { labels: [Label { name: "__name__", value: "test_summary" }, Label { name: "quantile", value: "100" }], samples: [Sample { value: 5.0, timestamp: 20 }], exemplars: [], histograms: [] }
+TimeSeries { labels: [Label { name: "__name__", value: "test_summary_sum" }], samples: [Sample { value: 15.0, timestamp: 20 }], exemplars: [], histograms: [] }
+TimeSeries { labels: [Label { name: "__name__", value: "test_summary_count" }], samples: [Sample { value: 5.0, timestamp: 20 }], exemplars: [], histograms: [] }"#;
         assert_eq!(write_quest_str.join("\n"), ans);
     }
 
@@ -385,11 +394,11 @@ TimeSeries { labels: [Label { name: "__name__", value: "test_summary_count" }],
         let write_quest2 = convert_metric_to_write_request(mf, Some(&filter), 0);
         assert_eq!(
             format!("{:?}", write_quest1.timeseries),
-            r#"[TimeSeries { labels: [Label { name: "__name__", value: "filter_counter" }, Label { name: "a", value: "1" }, Label { name: "b", value: "2" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [] }, TimeSeries { labels: [Label { name: "__name__", value: "test_counter" }, Label { name: "a", value: "1" }, Label { name: "b", value: "2" }], samples: [Sample { value: 2.0, timestamp: 0 }], exemplars: [] }]"#
+            r#"[TimeSeries { labels: [Label { name: "__name__", value: "filter_counter" }, Label { name: "a", value: "1" }, Label { name: "b", value: "2" }], samples: [Sample { value: 1.0, timestamp: 0 }], exemplars: [], histograms: [] }, TimeSeries { labels: [Label { name: "__name__", value: "test_counter" }, Label { name: "a", value: "1" }, Label { name: "b", value: "2" }], samples: [Sample { value: 2.0, timestamp: 0 }], exemplars: [], histograms: [] }]"#
         );
         assert_eq!(
             format!("{:?}", write_quest2.timeseries),
-            r#"[TimeSeries { labels: [Label { name: "__name__", value: "test_counter" }, Label { name: "a", value: "1" }, Label { name: "b", value: "2" }], samples: [Sample { value: 2.0, timestamp: 0 }], exemplars: [] }]"#
+            r#"[TimeSeries { labels: [Label { name: "__name__", value: "test_counter" }, Label { name: "a", value: "1" }, Label { name: "b", value: "2" }], samples: [Sample { value: 2.0, timestamp: 0 }], exemplars: [], histograms: [] }]"#
         );
     }
 }

src/common/wal/src/config.rs

@@ -206,6 +206,8 @@ mod tests {
                     client_cert_path: None,
                     client_key_path: None,
                 }),
+                connect_timeout: Duration::from_secs(3),
+                timeout: Duration::from_secs(3),
             },
             kafka_topic: KafkaTopicConfig {
                 num_topics: 32,
@@ -239,6 +241,8 @@ mod tests {
                     client_cert_path: None,
                     client_key_path: None,
                 }),
+                connect_timeout: Duration::from_secs(3),
+                timeout: Duration::from_secs(3),
             },
             max_batch_bytes: ReadableSize::mb(1),
             consumer_wait_timeout: Duration::from_millis(100),

src/common/wal/src/config/kafka/common.rs

@@ -36,9 +36,6 @@ pub const DEFAULT_BACKOFF_CONFIG: BackoffConfig = BackoffConfig {
     deadline: Some(Duration::from_secs(3)),
 };
 
-/// The default connect timeout for kafka client.
-pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
-
 /// Default interval for auto WAL pruning.
 pub const DEFAULT_AUTO_PRUNE_INTERVAL: Duration = Duration::from_mins(30);
 /// Default limit for concurrent auto pruning tasks.
@@ -167,6 +164,12 @@ pub struct KafkaConnectionConfig {
     pub sasl: Option<KafkaClientSasl>,
     /// Client TLS config
     pub tls: Option<KafkaClientTls>,
+    /// The connect timeout for kafka client.
+    #[serde(with = "humantime_serde")]
+    pub connect_timeout: Duration,
+    /// The timeout for kafka client.
+    #[serde(with = "humantime_serde")]
+    pub timeout: Duration,
 }
 
 impl Default for KafkaConnectionConfig {
@@ -175,6 +178,8 @@ impl Default for KafkaConnectionConfig {
             broker_endpoints: vec![BROKER_ENDPOINT.to_string()],
             sasl: None,
             tls: None,
+            connect_timeout: Duration::from_secs(3),
+            timeout: Duration::from_secs(3),
         }
     }
 }

src/frontend/src/frontend.rs

@@ -157,7 +157,6 @@ mod tests {
     use common_error::from_header_to_err_code_msg;
     use common_error::status_code::StatusCode;
     use common_grpc::channel_manager::ChannelManager;
-    use common_meta::distributed_time_constants::FRONTEND_HEARTBEAT_INTERVAL_MILLIS;
     use common_meta::heartbeat::handler::HandlerGroupExecutor;
     use common_meta::heartbeat::handler::parse_mailbox_message::ParseMailboxMessageHandler;
     use common_meta::heartbeat::handler::suspend::SuspendHandler;
@@ -400,6 +399,10 @@ mod tests {
                 ..Default::default()
             },
             meta_client: Some(meta_client_options.clone()),
+            heartbeat: HeartbeatOptions {
+                interval: Duration::from_secs(1),
+                ..Default::default()
+            },
             ..Default::default()
         };
 
@@ -409,7 +412,8 @@ mod tests {
         let meta_client = create_meta_client(&meta_client_options, server.clone()).await;
         let frontend = create_frontend(&options, meta_client).await?;
 
-        tokio::time::sleep(Duration::from_millis(FRONTEND_HEARTBEAT_INTERVAL_MILLIS)).await;
+        let frontend_heartbeat_interval = options.heartbeat.interval;
+        tokio::time::sleep(frontend_heartbeat_interval).await;
         // initial state: not suspend:
         assert!(!frontend.instance.is_suspended());
         verify_suspend_state_by_http(&frontend, Ok(r#"[{"records":{"schema":{"column_schemas":[{"name":"Int64(1)","data_type":"Int64"}]},"rows":[[1]],"total_rows":1}}]"#)).await;
@@ -426,7 +430,7 @@ mod tests {
 
         // make heartbeat server returned "suspend" instruction,
         server.suspend.store(true, Ordering::Relaxed);
-        tokio::time::sleep(Duration::from_millis(FRONTEND_HEARTBEAT_INTERVAL_MILLIS)).await;
+        tokio::time::sleep(frontend_heartbeat_interval).await;
         // ... then the frontend is suspended:
         assert!(frontend.instance.is_suspended());
         verify_suspend_state_by_http(
@@ -442,7 +446,7 @@ mod tests {
 
         // make heartbeat server NOT returned "suspend" instruction,
         server.suspend.store(false, Ordering::Relaxed);
-        tokio::time::sleep(Duration::from_millis(FRONTEND_HEARTBEAT_INTERVAL_MILLIS)).await;
+        tokio::time::sleep(frontend_heartbeat_interval).await;
         // ... then frontend's suspend state is cleared:
         assert!(!frontend.instance.is_suspended());
         verify_suspend_state_by_http(&frontend, Ok(r#"[{"records":{"schema":{"column_schemas":[{"name":"Int64(1)","data_type":"Int64"}]},"rows":[[1]],"total_rows":1}}]"#)).await;

src/log-store/src/kafka/client_manager.rs

@@ -16,7 +16,7 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use common_wal::config::kafka::DatanodeKafkaConfig;
-use common_wal::config::kafka::common::{DEFAULT_BACKOFF_CONFIG, DEFAULT_CONNECT_TIMEOUT};
+use common_wal::config::kafka::common::DEFAULT_BACKOFF_CONFIG;
 use dashmap::DashMap;
 use rskafka::client::ClientBuilder;
 use rskafka::client::partition::{Compression, PartitionClient, UnknownTopicHandling};
@@ -79,7 +79,8 @@ impl ClientManager {
         // Sets backoff config for the top-level kafka client and all clients constructed by it.
         let mut builder = ClientBuilder::new(config.connection.broker_endpoints.clone())
             .backoff_config(DEFAULT_BACKOFF_CONFIG)
-            .connect_timeout(Some(DEFAULT_CONNECT_TIMEOUT));
+            .connect_timeout(Some(config.connection.connect_timeout))
+            .timeout(Some(config.connection.timeout));
         if let Some(sasl) = &config.connection.sasl {
             builder = builder.sasl_config(sasl.config.clone().into_sasl_config());
         };

src/meta-srv/src/bootstrap.rs

@@ -14,7 +14,6 @@
 
 use std::net::SocketAddr;
 use std::sync::Arc;
-use std::time::Duration;
 
 use api::v1::meta::cluster_server::ClusterServer;
 use api::v1::meta::heartbeat_server::HeartbeatServer;
@@ -60,11 +59,6 @@ use crate::service::admin::admin_axum_router;
 use crate::utils::etcd::create_etcd_client_with_tls;
 use crate::{Result, error};
 
-/// The default keep-alive interval for gRPC.
-const DEFAULT_GRPC_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(10);
-/// The default keep-alive timeout for gRPC.
-const DEFAULT_GRPC_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(10);
-
 pub struct MetasrvInstance {
     metasrv: Arc<Metasrv>,
 
@@ -255,8 +249,8 @@ pub fn router(metasrv: Arc<Metasrv>) -> Router {
         // for admin services
         .accept_http1(true)
         // For quick network failures detection.
-        .http2_keepalive_interval(Some(DEFAULT_GRPC_KEEP_ALIVE_INTERVAL))
-        .http2_keepalive_timeout(Some(DEFAULT_GRPC_KEEP_ALIVE_TIMEOUT));
+        .http2_keepalive_interval(Some(metasrv.options().grpc.http2_keep_alive_interval))
+        .http2_keepalive_timeout(Some(metasrv.options().grpc.http2_keep_alive_timeout));
     let router = add_compressed_service!(router, HeartbeatServer::from_arc(metasrv.clone()));
     let router = add_compressed_service!(router, StoreServer::from_arc(metasrv.clone()));
     let router = add_compressed_service!(router, ClusterServer::from_arc(metasrv.clone()));
@@ -273,8 +267,12 @@ pub async fn metasrv_builder(
         (Some(kv_backend), _) => (kv_backend, None),
         (None, BackendImpl::MemoryStore) => (Arc::new(MemoryKvBackend::new()) as _, None),
         (None, BackendImpl::EtcdStore) => {
-            let etcd_client =
-                create_etcd_client_with_tls(&opts.store_addrs, opts.backend_tls.as_ref()).await?;
+            let etcd_client = create_etcd_client_with_tls(
+                &opts.store_addrs,
+                &opts.backend_client,
+                opts.backend_tls.as_ref(),
+            )
+            .await?;
             let kv_backend = EtcdStore::with_etcd_client(etcd_client.clone(), opts.max_txn_ops);
             let election = EtcdElection::with_etcd_client(
                 &opts.grpc.server_addr,

src/meta-srv/src/discovery.rs

@@ -16,13 +16,9 @@ pub mod lease;
 pub mod node_info;
 pub mod utils;
 
-use std::time::Duration;
-
 use api::v1::meta::heartbeat_request::NodeWorkloads;
 use common_error::ext::BoxedError;
-use common_meta::distributed_time_constants::{
-    DATANODE_LEASE_SECS, FLOWNODE_LEASE_SECS, FRONTEND_HEARTBEAT_INTERVAL_MILLIS,
-};
+use common_meta::distributed_time_constants::default_distributed_time_constants;
 use common_meta::error::Result;
 use common_meta::peer::{Peer, PeerDiscovery, PeerResolver};
 use common_meta::{DatanodeId, FlownodeId};
@@ -38,7 +34,7 @@ impl PeerDiscovery for MetaPeerClient {
         utils::alive_frontends(
             &DefaultSystemTimer,
             self,
-            Duration::from_millis(FRONTEND_HEARTBEAT_INTERVAL_MILLIS),
+            default_distributed_time_constants().frontend_heartbeat_interval,
         )
         .await
         .map_err(BoxedError::new)
@@ -52,7 +48,7 @@ impl PeerDiscovery for MetaPeerClient {
         utils::alive_datanodes(
             &DefaultSystemTimer,
             self,
-            Duration::from_secs(DATANODE_LEASE_SECS),
+            default_distributed_time_constants().datanode_lease,
             filter,
         )
         .await
@@ -67,7 +63,7 @@ impl PeerDiscovery for MetaPeerClient {
         utils::alive_flownodes(
             &DefaultSystemTimer,
             self,
-            Duration::from_secs(FLOWNODE_LEASE_SECS),
+            default_distributed_time_constants().flownode_lease,
             filter,
         )
         .await

src/meta-srv/src/discovery/lease.rs

@@ -102,7 +102,7 @@ mod tests {
     use api::v1::meta::heartbeat_request::NodeWorkloads;
     use api::v1::meta::{DatanodeWorkloads, FlownodeWorkloads};
     use common_meta::cluster::{FrontendStatus, NodeInfo, NodeInfoKey, NodeStatus, Role};
-    use common_meta::distributed_time_constants::FRONTEND_HEARTBEAT_INTERVAL_MILLIS;
+    use common_meta::distributed_time_constants::default_distributed_time_constants;
     use common_meta::kv_backend::ResettableKvBackendRef;
     use common_meta::peer::{Peer, PeerDiscovery};
     use common_meta::rpc::store::PutRequest;
@@ -473,8 +473,10 @@ mod tests {
         let client = create_meta_peer_client();
         let in_memory = client.memory_backend();
 
+        let frontend_heartbeat_interval =
+            default_distributed_time_constants().frontend_heartbeat_interval;
         let last_activity_ts =
-            current_time_millis() - FRONTEND_HEARTBEAT_INTERVAL_MILLIS as i64 - 1000;
+            current_time_millis() - frontend_heartbeat_interval.as_millis() as i64 - 1000;
         let active_frontend_node = NodeInfo {
             peer: Peer {
                 id: 0,

src/meta-srv/src/failure_detector.rs

@@ -15,7 +15,6 @@
 use std::collections::VecDeque;
 use std::time::Duration;
 
-use common_meta::distributed_time_constants;
 use serde::{Deserialize, Serialize};
 
 const FIRST_HEARTBEAT_ESTIMATE_MILLIS: i64 = 1000;
@@ -79,9 +78,7 @@ impl Default for PhiAccrualFailureDetectorOptions {
         Self {
             threshold: 8_f32,
             min_std_deviation: Duration::from_millis(100),
-            acceptable_heartbeat_pause: Duration::from_secs(
-                distributed_time_constants::DATANODE_LEASE_SECS,
-            ),
+            acceptable_heartbeat_pause: Duration::from_secs(10),
         }
     }
 }

src/meta-srv/src/handler/region_lease_handler.rs

@@ -134,7 +134,7 @@ mod test {
     use std::sync::Arc;
 
     use common_meta::datanode::{RegionManifestInfo, RegionStat, Stat};
-    use common_meta::distributed_time_constants;
+    use common_meta::distributed_time_constants::default_distributed_time_constants;
     use common_meta::key::TableMetadataManager;
     use common_meta::key::table_route::TableRouteValue;
     use common_meta::key::test_utils::new_test_table_info;
@@ -236,7 +236,7 @@ mod test {
         let opening_region_keeper = Arc::new(MemoryRegionKeeper::default());
 
         let handler = RegionLeaseHandler::new(
-            distributed_time_constants::REGION_LEASE_SECS,
+            default_distributed_time_constants().region_lease.as_secs(),
             table_metadata_manager.clone(),
             opening_region_keeper.clone(),
             None,
@@ -266,7 +266,7 @@ mod test {
 
         assert_eq!(
             acc.region_lease.as_ref().unwrap().lease_seconds,
-            distributed_time_constants::REGION_LEASE_SECS
+            default_distributed_time_constants().region_lease.as_secs()
         );
 
         assert_region_lease(
@@ -300,7 +300,7 @@ mod test {
 
         assert_eq!(
             acc.region_lease.as_ref().unwrap().lease_seconds,
-            distributed_time_constants::REGION_LEASE_SECS
+            default_distributed_time_constants().region_lease.as_secs()
         );
 
         assert_region_lease(
@@ -379,7 +379,7 @@ mod test {
         });
 
         let handler = RegionLeaseHandler::new(
-            distributed_time_constants::REGION_LEASE_SECS,
+            default_distributed_time_constants().region_lease.as_secs(),
             table_metadata_manager.clone(),
             Default::default(),
             None,
@@ -461,7 +461,7 @@ mod test {
             ..Default::default()
         });
         let handler = RegionLeaseHandler::new(
-            distributed_time_constants::REGION_LEASE_SECS,
+            default_distributed_time_constants().region_lease.as_secs(),
             table_metadata_manager.clone(),
             Default::default(),
             None,

src/meta-srv/src/metasrv.rs

@@ -27,7 +27,7 @@ use common_event_recorder::EventRecorderOptions;
 use common_greptimedb_telemetry::GreptimeDBTelemetryTask;
 use common_meta::cache_invalidator::CacheInvalidatorRef;
 use common_meta::ddl_manager::DdlManagerRef;
-use common_meta::distributed_time_constants;
+use common_meta::distributed_time_constants::{self, default_distributed_time_constants};
 use common_meta::key::TableMetadataManagerRef;
 use common_meta::key::runtime_switch::RuntimeSwitchManagerRef;
 use common_meta::kv_backend::{KvBackendRef, ResettableKvBackend, ResettableKvBackendRef};
@@ -121,6 +121,27 @@ impl Default for StatsPersistenceOptions {
     }
 }
 
+#[derive(Clone, PartialEq, Serialize, Deserialize, Debug)]
+#[serde(default)]
+pub struct BackendClientOptions {
+    #[serde(with = "humantime_serde")]
+    pub keep_alive_timeout: Duration,
+    #[serde(with = "humantime_serde")]
+    pub keep_alive_interval: Duration,
+    #[serde(with = "humantime_serde")]
+    pub connect_timeout: Duration,
+}
+
+impl Default for BackendClientOptions {
+    fn default() -> Self {
+        Self {
+            keep_alive_interval: Duration::from_secs(10),
+            keep_alive_timeout: Duration::from_secs(3),
+            connect_timeout: Duration::from_secs(3),
+        }
+    }
+}
+
 #[derive(Clone, PartialEq, Serialize, Deserialize)]
 #[serde(default)]
 pub struct MetasrvOptions {
@@ -136,12 +157,20 @@ pub struct MetasrvOptions {
     /// Only applicable when using PostgreSQL or MySQL as the metadata store
     #[serde(default)]
     pub backend_tls: Option<TlsOption>,
+    /// The backend client options.
+    /// Currently, only applicable when using etcd as the metadata store.
+    #[serde(default)]
+    pub backend_client: BackendClientOptions,
     /// The type of selector.
     pub selector: SelectorType,
-    /// Whether to use the memory store.
-    pub use_memory_store: bool,
     /// Whether to enable region failover.
     pub enable_region_failover: bool,
+    /// The base heartbeat interval.
+    ///
+    /// This value is used to calculate the distributed time constants for components.
+    /// e.g., the region lease time is `heartbeat_interval * 3 + Duration::from_secs(1)`.
+    #[serde(with = "humantime_serde")]
+    pub heartbeat_interval: Duration,
     /// The delay before starting region failure detection.
     /// This delay helps prevent Metasrv from triggering unnecessary region failovers before all Datanodes are fully started.
     /// Especially useful when the cluster is not deployed with GreptimeDB Operator and maintenance mode is not enabled.
@@ -219,7 +248,6 @@ impl fmt::Debug for MetasrvOptions {
             .field("store_addrs", &self.sanitize_store_addrs())
             .field("backend_tls", &self.backend_tls)
             .field("selector", &self.selector)
-            .field("use_memory_store", &self.use_memory_store)
             .field("enable_region_failover", &self.enable_region_failover)
             .field(
                 "allow_region_failover_on_local_wal",
@@ -240,7 +268,9 @@ impl fmt::Debug for MetasrvOptions {
             .field("tracing", &self.tracing)
             .field("backend", &self.backend)
             .field("event_recorder", &self.event_recorder)
-            .field("stats_persistence", &self.stats_persistence);
+            .field("stats_persistence", &self.stats_persistence)
+            .field("heartbeat_interval", &self.heartbeat_interval)
+            .field("backend_client", &self.backend_client);
 
         #[cfg(any(feature = "pg_kvbackend", feature = "mysql_kvbackend"))]
         debug_struct.field("meta_table_name", &self.meta_table_name);
@@ -268,8 +298,8 @@ impl Default for MetasrvOptions {
             store_addrs: vec!["127.0.0.1:2379".to_string()],
             backend_tls: None,
             selector: SelectorType::default(),
-            use_memory_store: false,
             enable_region_failover: false,
+            heartbeat_interval: distributed_time_constants::BASE_HEARTBEAT_INTERVAL,
             region_failure_detector_initialization_delay: Duration::from_secs(10 * 60),
             allow_region_failover_on_local_wal: false,
             grpc: GrpcOptions {
@@ -307,6 +337,7 @@ impl Default for MetasrvOptions {
             event_recorder: EventRecorderOptions::default(),
             stats_persistence: StatsPersistenceOptions::default(),
             gc: GcSchedulerOptions::default(),
+            backend_client: BackendClientOptions::default(),
         }
     }
 }
@@ -747,7 +778,7 @@ impl Metasrv {
             &DefaultSystemTimer,
             self.meta_peer_client.as_ref(),
             peer_id,
-            Duration::from_secs(distributed_time_constants::DATANODE_LEASE_SECS),
+            default_distributed_time_constants().datanode_lease,
         )
         .await
     }

src/meta-srv/src/metasrv/builder.rs

@@ -29,7 +29,7 @@ use common_meta::ddl::{
     DdlContext, NoopRegionFailureDetectorControl, RegionFailureDetectorControllerRef,
 };
 use common_meta::ddl_manager::{DdlManager, DdlManagerConfiguratorRef};
-use common_meta::distributed_time_constants::{self};
+use common_meta::distributed_time_constants::default_distributed_time_constants;
 use common_meta::key::TableMetadataManager;
 use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::flow::flow_state::FlowStateManager;
@@ -513,7 +513,7 @@ impl MetasrvBuilder {
             Some(handler_group_builder) => handler_group_builder,
             None => {
                 let region_lease_handler = RegionLeaseHandler::new(
-                    distributed_time_constants::REGION_LEASE_SECS,
+                    default_distributed_time_constants().region_lease.as_secs(),
                     table_metadata_manager.clone(),
                     memory_region_keeper.clone(),
                     customized_region_lease_renewer,

src/meta-srv/src/procedure/region_migration.rs

@@ -921,7 +921,7 @@ mod tests {
     use std::assert_matches::assert_matches;
     use std::sync::Arc;
 
-    use common_meta::distributed_time_constants::REGION_LEASE_SECS;
+    use common_meta::distributed_time_constants::default_distributed_time_constants;
     use common_meta::instruction::Instruction;
     use common_meta::key::test_utils::new_test_table_info;
     use common_meta::rpc::router::{Region, RegionRoute};
@@ -1192,8 +1192,10 @@ mod tests {
             .run_once()
             .await;
 
+        let region_lease = default_distributed_time_constants().region_lease.as_secs();
+
         // Ensure it didn't run into the slow path.
-        assert!(timer.elapsed().as_secs() < REGION_LEASE_SECS / 2);
+        assert!(timer.elapsed().as_secs() < region_lease / 2);
 
         runner.suite.verify_table_metadata().await;
     }
@@ -1539,8 +1541,9 @@ mod tests {
             .run_once()
             .await;
 
+        let region_lease = default_distributed_time_constants().region_lease.as_secs();
         // Ensure it didn't run into the slow path.
-        assert!(timer.elapsed().as_secs() < REGION_LEASE_SECS);
+        assert!(timer.elapsed().as_secs() < region_lease);
         runner.suite.verify_table_metadata().await;
     }
 }

src/meta-srv/src/procedure/region_migration/close_downgraded_region.rs

@@ -13,11 +13,10 @@
 // limitations under the License.
 
 use std::any::Any;
-use std::time::Duration;
 
 use api::v1::meta::MailboxMessage;
 use common_meta::RegionIdent;
-use common_meta::distributed_time_constants::REGION_LEASE_SECS;
+use common_meta::distributed_time_constants::default_distributed_time_constants;
 use common_meta::instruction::{Instruction, InstructionReply, SimpleReply};
 use common_procedure::{Context as ProcedureContext, Status};
 use common_telemetry::{info, warn};
@@ -30,9 +29,6 @@ use crate::procedure::region_migration::migration_end::RegionMigrationEnd;
 use crate::procedure::region_migration::{Context, State};
 use crate::service::mailbox::Channel;
 
-/// Uses lease time of a region as the timeout of closing a downgraded region.
-const CLOSE_DOWNGRADED_REGION_TIMEOUT: Duration = Duration::from_secs(REGION_LEASE_SECS);
-
 #[derive(Debug, Serialize, Deserialize)]
 pub struct CloseDowngradedRegion;
 
@@ -112,7 +108,7 @@ impl CloseDowngradedRegion {
         let ch = Channel::Datanode(downgrade_leader_datanode.id);
         let receiver = ctx
             .mailbox
-            .send(&ch, msg, CLOSE_DOWNGRADED_REGION_TIMEOUT)
+            .send(&ch, msg, default_distributed_time_constants().region_lease)
             .await?;
 
         match receiver.await {

src/meta-srv/src/procedure/region_migration/downgrade_leader_region.rs

@@ -17,7 +17,7 @@ use std::time::Duration;
 
 use api::v1::meta::MailboxMessage;
 use common_error::ext::BoxedError;
-use common_meta::distributed_time_constants::REGION_LEASE_SECS;
+use common_meta::distributed_time_constants::default_distributed_time_constants;
 use common_meta::instruction::{
     DowngradeRegion, DowngradeRegionReply, DowngradeRegionsReply, Instruction, InstructionReply,
 };
@@ -64,7 +64,7 @@ impl State for DowngradeLeaderRegion {
         let now = Instant::now();
         // Ensures the `leader_region_lease_deadline` must exist after recovering.
         ctx.volatile_ctx
-            .set_leader_region_lease_deadline(Duration::from_secs(REGION_LEASE_SECS));
+            .set_leader_region_lease_deadline(default_distributed_time_constants().region_lease);
 
         match self.downgrade_region_with_retry(ctx).await {
             Ok(_) => {
@@ -277,14 +277,14 @@ impl DowngradeLeaderRegion {
         if let Some(last_connection_at) = last_connection_at {
             let now = current_time_millis();
             let elapsed = now - last_connection_at;
-            let region_lease = Duration::from_secs(REGION_LEASE_SECS);
+            let region_lease = default_distributed_time_constants().region_lease;
 
             // It's safe to update the region leader lease deadline here because:
             // 1. The old region leader has already been marked as downgraded in metadata,
             //    which means any attempts to renew its lease will be rejected.
             // 2. The pusher disconnect time record only gets removed when the datanode (from_peer)
             //    establishes a new heartbeat connection stream.
-            if elapsed >= (REGION_LEASE_SECS * 1000) as i64 {
+            if elapsed >= (region_lease.as_secs() * 1000) as i64 {
                 ctx.volatile_ctx.reset_leader_region_lease_deadline();
                 info!(
                     "Datanode {}({}) has been disconnected for longer than the region lease period ({:?}), reset leader region lease deadline to None, region: {:?}",
@@ -697,7 +697,8 @@ mod tests {
         let procedure_ctx = new_procedure_context();
         let (next, _) = state.next(&mut ctx, &procedure_ctx).await.unwrap();
         let elapsed = timer.elapsed().as_secs();
-        assert!(elapsed < REGION_LEASE_SECS / 2);
+        let region_lease = default_distributed_time_constants().region_lease.as_secs();
+        assert!(elapsed < region_lease / 2);
         assert_eq!(
             ctx.volatile_ctx
                 .leader_region_last_entry_ids

src/meta-srv/src/procedure/region_migration/open_candidate_region.rs

@@ -14,11 +14,10 @@
 
 use std::any::Any;
 use std::ops::Div;
-use std::time::Duration;
 
 use api::v1::meta::MailboxMessage;
 use common_meta::RegionIdent;
-use common_meta::distributed_time_constants::REGION_LEASE_SECS;
+use common_meta::distributed_time_constants::default_distributed_time_constants;
 use common_meta::instruction::{Instruction, InstructionReply, OpenRegion, SimpleReply};
 use common_meta::key::datanode_table::RegionInfo;
 use common_procedure::{Context as ProcedureContext, Status};
@@ -33,9 +32,6 @@ use crate::procedure::region_migration::flush_leader_region::PreFlushRegion;
 use crate::procedure::region_migration::{Context, State};
 use crate::service::mailbox::Channel;
 
-/// Uses lease time of a region as the timeout of opening a candidate region.
-const OPEN_CANDIDATE_REGION_TIMEOUT: Duration = Duration::from_secs(REGION_LEASE_SECS);
-
 #[derive(Debug, Serialize, Deserialize)]
 pub struct OpenCandidateRegion;
 
@@ -157,7 +153,9 @@ impl OpenCandidateRegion {
                 .context(error::ExceededDeadlineSnafu {
                     operation: "Open candidate region",
                 })?;
-        let operation_timeout = operation_timeout.div(2).max(OPEN_CANDIDATE_REGION_TIMEOUT);
+        let operation_timeout = operation_timeout
+            .div(2)
+            .max(default_distributed_time_constants().region_lease);
         let ch = Channel::Datanode(candidate.id);
         let now = Instant::now();
         let receiver = ctx.mailbox.send(&ch, msg, operation_timeout).await?;

src/meta-srv/src/service/heartbeat.rs

@@ -99,6 +99,7 @@ impl heartbeat_server::Heartbeat for Metasrv {
                             error!("Client disconnected: broken pipe");
                             break;
                         }
+                        error!(err; "Sending heartbeat response error");
 
                         if tx.send(Err(err)).await.is_err() {
                             info!("ReceiverStream was dropped; shutting down");

src/meta-srv/src/utils/etcd.rs

@@ -12,17 +12,18 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use common_meta::distributed_time_constants::default_etcd_client_options;
 use common_meta::kv_backend::etcd::create_etcd_tls_options;
-use etcd_client::Client;
+use etcd_client::{Client, ConnectOptions};
 use servers::tls::{TlsMode, TlsOption};
 use snafu::ResultExt;
 
 use crate::error::{self, BuildTlsOptionsSnafu, Result};
+use crate::metasrv::BackendClientOptions;
 
 /// Creates an etcd client with TLS configuration.
 pub async fn create_etcd_client_with_tls(
     store_addrs: &[String],
+    client_options: &BackendClientOptions,
     tls_config: Option<&TlsOption>,
 ) -> Result<Client> {
     let etcd_endpoints = store_addrs
@@ -31,7 +32,12 @@ pub async fn create_etcd_client_with_tls(
         .filter(|x| !x.is_empty())
         .collect::<Vec<_>>();
 
-    let mut connect_options = default_etcd_client_options();
+    let mut connect_options = ConnectOptions::new()
+        .with_keep_alive_while_idle(true)
+        .with_keep_alive(
+            client_options.keep_alive_interval,
+            client_options.keep_alive_timeout,
+        );
     if let Some(tls_config) = tls_config
         && let Some(tls_options) = create_etcd_tls_options(&convert_tls_option(tls_config))
             .context(BuildTlsOptionsSnafu)?

src/mito2/src/gc.rs

@@ -540,7 +540,7 @@ impl LocalGcWorker {
     fn filter_deletable_files(
         &self,
         entries: Vec<Entry>,
-        in_use_filenames: &HashSet<&FileId>,
+        in_use_filenames: &HashSet<FileId>,
         may_linger_filenames: &HashSet<&FileId>,
         eligible_for_removal: &HashSet<&FileId>,
         unknown_file_may_linger_until: chrono::DateTime<chrono::Utc>,
@@ -641,9 +641,6 @@ impl LocalGcWorker {
             .flatten()
             .collect::<HashSet<_>>();
 
-        // in use filenames, include sst and index files
-        let in_use_filenames = in_used.iter().collect::<HashSet<_>>();
-
         // When full_file_listing is false, skip expensive list operations and only delete
         // files that are tracked in recently_removed_files
         if !self.full_file_listing {
@@ -653,7 +650,7 @@ impl LocalGcWorker {
             // 3. Have passed the lingering time
             let files_to_delete: Vec<FileId> = eligible_for_removal
                 .iter()
-                .filter(|file_id| !in_use_filenames.contains(*file_id))
+                .filter(|file_id| !in_used.contains(*file_id))
                 .map(|&f| *f)
                 .collect();
 
@@ -672,7 +669,7 @@ impl LocalGcWorker {
         let (all_unused_files_ready_for_delete, all_in_exist_linger_files) = self
             .filter_deletable_files(
                 all_entries,
-                &in_use_filenames,
+                in_used,
                 &may_linger_filenames,
                 &eligible_for_removal,
                 unknown_file_may_linger_until,

src/object-store/src/config.rs

@@ -16,7 +16,6 @@ use std::time::Duration;
 
 use common_base::readable_size::ReadableSize;
 use common_base::secrets::{ExposeSecret, SecretString};
-use common_telemetry::tracing::warn;
 use opendal::services::{Azblob, Gcs, Oss, S3};
 use serde::{Deserialize, Serialize};
 
@@ -118,23 +117,25 @@ pub struct S3Connection {
     /// By default, opendal will send API to https://s3.us-east-1.amazonaws.com/bucket_name
     /// Enabled, opendal will send API to https://bucket_name.s3.us-east-1.amazonaws.com
     pub enable_virtual_host_style: bool,
+    /// Disable EC2 metadata service.
+    /// By default, opendal will use EC2 metadata service to load credentials from the instance metadata,
+    /// when access key id and secret access key are not provided.
+    /// If enabled, opendal will *NOT* use EC2 metadata service.
+    pub disable_ec2_metadata: bool,
 }
 
 impl From<&S3Connection> for S3 {
     fn from(connection: &S3Connection) -> Self {
         let root = util::normalize_dir(&connection.root);
 
-        let mut builder = S3::default().root(&root).bucket(&connection.bucket);
+        let mut builder = S3::default()
+            .root(&root)
+            .bucket(&connection.bucket)
+            .access_key_id(connection.access_key_id.expose_secret())
+            .secret_access_key(connection.secret_access_key.expose_secret());
 
-        if !connection.access_key_id.expose_secret().is_empty()
-            && !connection.secret_access_key.expose_secret().is_empty()
-        {
-            builder = builder
-                .access_key_id(connection.access_key_id.expose_secret())
-                .secret_access_key(connection.secret_access_key.expose_secret());
-        } else {
-            warn!("No access key id or secret access key provided, using anonymous access");
-            builder = builder.allow_anonymous().disable_ec2_metadata();
+        if connection.disable_ec2_metadata {
+            builder = builder.disable_ec2_metadata();
         }
 
         if let Some(endpoint) = &connection.endpoint {

src/servers/src/grpc.rs

@@ -81,6 +81,12 @@ pub struct GrpcOptions {
     /// Default to `None`, means infinite.
     #[serde(with = "humantime_serde")]
     pub max_connection_age: Option<Duration>,
+    /// The HTTP/2 keep-alive interval.
+    #[serde(with = "humantime_serde")]
+    pub http2_keep_alive_interval: Duration,
+    /// The HTTP/2 keep-alive timeout.
+    #[serde(with = "humantime_serde")]
+    pub http2_keep_alive_timeout: Duration,
 }
 
 impl GrpcOptions {
@@ -144,6 +150,8 @@ impl Default for GrpcOptions {
             runtime_size: 8,
             tls: TlsOption::default(),
             max_connection_age: None,
+            http2_keep_alive_interval: Duration::from_secs(10),
+            http2_keep_alive_timeout: Duration::from_secs(3),
         }
     }
 }
@@ -164,6 +172,8 @@ impl GrpcOptions {
             runtime_size: 8,
             tls: TlsOption::default(),
             max_connection_age: None,
+            http2_keep_alive_interval: Duration::from_secs(10),
+            http2_keep_alive_timeout: Duration::from_secs(3),
         }
     }
 

src/servers/src/heartbeat_options.rs

@@ -34,12 +34,10 @@ impl HeartbeatOptions {
     pub fn frontend_default() -> Self {
         Self {
             // Frontend can send heartbeat with a longer interval.
-            interval: Duration::from_millis(
-                distributed_time_constants::FRONTEND_HEARTBEAT_INTERVAL_MILLIS,
-            ),
-            retry_interval: Duration::from_millis(
-                distributed_time_constants::HEARTBEAT_INTERVAL_MILLIS,
+            interval: distributed_time_constants::frontend_heartbeat_interval(
+                distributed_time_constants::BASE_HEARTBEAT_INTERVAL,
             ),
+            retry_interval: distributed_time_constants::BASE_HEARTBEAT_INTERVAL,
         }
     }
 }
@@ -47,10 +45,8 @@ impl HeartbeatOptions {
 impl Default for HeartbeatOptions {
     fn default() -> Self {
         Self {
-            interval: Duration::from_millis(distributed_time_constants::HEARTBEAT_INTERVAL_MILLIS),
-            retry_interval: Duration::from_millis(
-                distributed_time_constants::HEARTBEAT_INTERVAL_MILLIS,
-            ),
+            interval: distributed_time_constants::BASE_HEARTBEAT_INTERVAL,
+            retry_interval: distributed_time_constants::BASE_HEARTBEAT_INTERVAL,
         }
     }
 }

src/servers/src/prom_store.rs

@@ -26,7 +26,7 @@ use arrow::datatypes::{Float64Type, TimestampMillisecondType};
 use common_grpc::precision::Precision;
 use common_query::prelude::{greptime_timestamp, greptime_value};
 use common_recordbatch::{RecordBatch, RecordBatches};
-use common_telemetry::tracing;
+use common_telemetry::{tracing, warn};
 use datafusion::dataframe::DataFrame;
 use datafusion::prelude::{Expr, col, lit, regexp_match};
 use datafusion_common::ScalarValue;
@@ -415,6 +415,10 @@ pub fn to_grpc_row_insert_requests(request: &WriteRequest) -> Result<(RowInsertR
                 table_data.add_row(one_row);
             }
         }
+
+        if !series.histograms.is_empty() {
+            warn!("Native histograms are not supported yet, data ignored");
+        }
     }
 
     Ok(multi_table_data.into_row_insert_requests())

src/servers/src/tls.rs

@@ -362,13 +362,13 @@ mod tests {
                 cert_path: "/path/to/cert_path".to_string(),
                 key_path: "/path/to/key_path".to_string(),
                 ca_cert_path: String::new(),
-                watch: false
+                watch: false,
             },
             TlsOption::new(
                 Some(Disable),
                 Some("/path/to/cert_path".to_string()),
                 Some("/path/to/key_path".to_string()),
-                false
+                false,
             )
         );
     }

tests-fuzz/targets/failover/fuzz_failover_metric_regions.rs

@@ -19,7 +19,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use arbitrary::{Arbitrary, Unstructured};
-use common_meta::distributed_time_constants;
+use common_meta::distributed_time_constants::default_distributed_time_constants;
 use common_telemetry::info;
 use libfuzzer_sys::fuzz_target;
 use rand::{Rng, SeedableRng};
@@ -254,10 +254,7 @@ async fn execute_failover(ctx: FuzzContext, input: FuzzInput) -> Result<()> {
     recover_pod_failure(ctx.kube.clone(), &ctx.namespace, &chaos_name).await?;
     wait_for_all_datanode_online(ctx.greptime.clone(), Duration::from_secs(60)).await;
 
-    tokio::time::sleep(Duration::from_secs(
-        distributed_time_constants::REGION_LEASE_SECS,
-    ))
-    .await;
+    tokio::time::sleep(default_distributed_time_constants().region_lease).await;
     // Validates value rows
     info!("Validates num of rows");
 

tests-fuzz/targets/failover/fuzz_failover_mito_regions.rs

@@ -19,7 +19,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use arbitrary::{Arbitrary, Unstructured};
-use common_meta::distributed_time_constants;
+use common_meta::distributed_time_constants::default_distributed_time_constants;
 use common_telemetry::info;
 use common_time::util::current_time_millis;
 use futures::future::try_join_all;
@@ -322,10 +322,7 @@ async fn execute_failover(ctx: FuzzContext, input: FuzzInput) -> Result<()> {
     recover_pod_failure(ctx.kube.clone(), &ctx.namespace, &chaos_name).await?;
     wait_for_all_datanode_online(ctx.greptime.clone(), Duration::from_secs(60)).await;
 
-    tokio::time::sleep(Duration::from_secs(
-        distributed_time_constants::REGION_LEASE_SECS,
-    ))
-    .await;
+    tokio::time::sleep(default_distributed_time_constants().region_lease).await;
     // Validates value rows
     info!("Validates num of rows");
     for (table_ctx, expected_rows) in table_ctxs.iter().zip(affected_rows) {

tests-fuzz/targets/migration/fuzz_migrate_metric_regions.rs

@@ -19,7 +19,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use arbitrary::{Arbitrary, Unstructured};
-use common_meta::distributed_time_constants;
+use common_meta::distributed_time_constants::default_distributed_time_constants;
 use common_telemetry::info;
 use libfuzzer_sys::fuzz_target;
 use rand::{Rng, SeedableRng};
@@ -275,10 +275,7 @@ async fn migrate_regions(ctx: &FuzzContext, migrations: &[Migration]) -> Result<
         wait_for_migration(ctx, migration, &procedure_id).await;
     }
 
-    tokio::time::sleep(Duration::from_secs(
-        distributed_time_constants::REGION_LEASE_SECS,
-    ))
-    .await;
+    tokio::time::sleep(default_distributed_time_constants().region_lease).await;
 
     Ok(())
 }

tests-fuzz/targets/migration/fuzz_migrate_mito_regions.rs

@@ -19,7 +19,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use arbitrary::{Arbitrary, Unstructured};
-use common_meta::distributed_time_constants;
+use common_meta::distributed_time_constants::default_distributed_time_constants;
 use common_telemetry::info;
 use libfuzzer_sys::fuzz_target;
 use rand::{Rng, SeedableRng};
@@ -274,10 +274,7 @@ async fn migrate_regions(ctx: &FuzzContext, migrations: &[Migration]) -> Result<
         .await;
     }
 
-    tokio::time::sleep(Duration::from_secs(
-        distributed_time_constants::REGION_LEASE_SECS,
-    ))
-    .await;
+    tokio::time::sleep(default_distributed_time_constants().region_lease).await;
 
     Ok(())
 }

tests-integration/tests/http.rs

@@ -1397,6 +1397,8 @@ max_recv_message_size = "512MiB"
 max_send_message_size = "512MiB"
 flight_compression = "arrow_ipc"
 runtime_size = 8
+http2_keep_alive_interval = "10s"
+http2_keep_alive_timeout = "3s"
 
 [grpc.tls]
 mode = "disable"
@@ -1586,6 +1588,7 @@ fn drop_lines_with_inconsistent_results(input: String) -> String {
         "endpoint =",
         "region =",
         "enable_virtual_host_style =",
+        "disable_ec2_metadata =",
         "cache_path =",
         "cache_capacity =",
         "memory_pool_size =",

tests/conf/metasrv-test.toml.template

@@ -3,9 +3,6 @@ flush_stats_factor = 1
 ## Store server address default to etcd store.
 store_addrs = [{store_addrs | unescaped}]
 
-## Store data in memory.
-use_memory_store = false
-
 ## The datastore for meta server.
 backend = "EtcdStore"
 {{ endif }}