fix: load batching protocols from environment and document frontend wiring

Signed-off-by: WenyXu <wenymedia@gmail.com>

Cargo.lock

@@ -5735,6 +5735,7 @@ dependencies = [
  "store-api",
  "strfmt",
  "table",
+ "temp-env",
  "tokio",
  "tokio-util",
  "toml 0.8.23",
@@ -14475,6 +14476,7 @@ dependencies = [
  "snafu 0.8.6",
  "store-api",
  "table",
+ "temp-env",
  "tokio",
  "toml 0.8.23",
  "url",

Cargo.toml

@@ -252,6 +252,7 @@ sqlx = { version = "0.8", default-features = false, features = [
 stringprep = "0.1"
 strum = { version = "0.27", features = ["derive"] }
 sysinfo = "0.33"
+temp-env = "0.3"
 tempfile = "3"
 tokio = { version = "1.47", features = ["full"] }
 tokio-postgres = "0.7.18"

config/config.md

@@ -33,7 +33,7 @@
 | `runtime.experimental_workload_scheduler.sample_every_polls` | Integer | `16` | Number of polls between scheduler fairness samples. Must be greater than zero. |
 | `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.<br/>When Prometheus pending-row batching is enabled, a nonzero timeout less than or equal to the<br/>`prom_store.pending_rows_flush_interval` plus 1 second is adjusted to that value. |
+| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.<br/>When synchronous Prometheus or shared table batching is enabled, a nonzero timeout is<br/>raised to at least the largest active flush interval plus 1 second. The intervals come from<br/>`prom_store.pending_rows_flush_interval` and `experimental_pending_rows_batcher.pending_rows_flush_interval`. |
 | `http.body_limit` | String | `64MB` | HTTP request body limit.<br/>The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.<br/>Set to 0 to disable limit. |
 | `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default<br/>This allows browser to access http APIs without CORS restrictions |
 | `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. |
@@ -286,7 +286,7 @@
 | `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 2). |
 | `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.<br/>When Prometheus pending-row batching is enabled, a nonzero timeout less than or equal to the<br/>`prom_store.pending_rows_flush_interval` plus 1 second is adjusted to that value. |
+| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.<br/>When synchronous Prometheus or shared table batching is enabled, a nonzero timeout is<br/>raised to at least the largest active flush interval plus 1 second. The intervals come from<br/>`prom_store.pending_rows_flush_interval` and `experimental_pending_rows_batcher.pending_rows_flush_interval`. |
 | `http.body_limit` | String | `64MB` | HTTP request body limit.<br/>The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.<br/>Set to 0 to disable limit. |
 | `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default<br/>This allows browser to access http APIs without CORS restrictions |
 | `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. |

config/frontend.example.toml

@@ -52,8 +52,9 @@ default_column_prefix = "greptime"
 ## The address to bind the HTTP server.
 addr = "127.0.0.1:4000"
 ## HTTP request timeout. Set to 0 to disable timeout.
-## When Prometheus pending-row batching is enabled, a nonzero timeout less than or equal to the
-## `prom_store.pending_rows_flush_interval` plus 1 second is adjusted to that value.
+## When synchronous Prometheus or shared table batching is enabled, a nonzero timeout is
+## raised to at least the largest active flush interval plus 1 second. The intervals come from
+## `prom_store.pending_rows_flush_interval` and `experimental_pending_rows_batcher.pending_rows_flush_interval`.
 timeout = "0s"
 ## HTTP request body limit.
 ## The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.

config/standalone.example.toml

@@ -77,8 +77,9 @@ max_concurrent_queries = 0
 ## The address to bind the HTTP server.
 addr = "127.0.0.1:4000"
 ## HTTP request timeout. Set to 0 to disable timeout.
-## When Prometheus pending-row batching is enabled, a nonzero timeout less than or equal to the
-## `prom_store.pending_rows_flush_interval` plus 1 second is adjusted to that value.
+## When synchronous Prometheus or shared table batching is enabled, a nonzero timeout is
+## raised to at least the largest active flush interval plus 1 second. The intervals come from
+## `prom_store.pending_rows_flush_interval` and `experimental_pending_rows_batcher.pending_rows_flush_interval`.
 timeout = "0s"
 ## HTTP request body limit.
 ## The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.

src/frontend/AGENTS.md

@@ -44,6 +44,12 @@ remote datanodes via `operator`/`client`.
   auto-create, partition routing) → local `RegionServer` (standalone) or RPC to
   datanodes (distributed).
 
+- **Table batching** (`instance/builder.rs`): protocol entry points opt in through
+  `QueryContext`. The primary inserter prepares eligible ordinary-table writes
+  for `servers::batcher::table::TablePendingRowsBatcher`. A separate execution-only
+  inserter, with no batcher attached, sends the prepared bulk writes to datanodes,
+  avoiding recursive batching. The batcher handles successful-write Flow notifications.
+
 ## Public surface
 
 - `Instance` (`instance.rs`) — the business-logic container.

src/frontend/Cargo.toml

@@ -89,6 +89,7 @@ tonic.workspace = true
 tracing.workspace = true
 
 [dev-dependencies]
+temp-env.workspace = true
 arrow-flight.workspace = true
 catalog = { workspace = true, features = ["testing"] }
 common-test-util.workspace = true

src/frontend/src/frontend.rs

@@ -127,6 +127,7 @@ impl Configurable for FrontendOptions {
             "heartbeat_env_vars",
             "meta_client.metasrv_addrs",
             "event_recorder.event_types",
+            "experimental_pending_rows_batcher.protocols",
         ])
     }
 }
@@ -221,6 +222,27 @@ mod tests {
     type GrpcStream<T> =
         Pin<Box<dyn Stream<Item = std::result::Result<T, Status>> + Send + Sync + 'static>>;
 
+    #[test]
+    fn test_batcher_protocols_from_env() {
+        temp_env::with_vars(
+            [(
+                "FRONTEND_BATCHER_TEST__EXPERIMENTAL_PENDING_ROWS_BATCHER__PROTOCOLS",
+                Some("influxdb,http_sql"),
+            )],
+            || {
+                let options =
+                    FrontendOptions::load_layered_options(None, "FRONTEND_BATCHER_TEST").unwrap();
+                assert_eq!(
+                    options.experimental_pending_rows_batcher.protocols,
+                    vec![
+                        servers::http::BatchingProtocol::Influxdb,
+                        servers::http::BatchingProtocol::HttpSql
+                    ]
+                );
+            },
+        );
+    }
+
     #[test]
     fn test_protocol_pending_rows_batcher_config() {
         let defaults: FrontendOptions = toml::from_str("").unwrap();

src/servers/src/http.rs

@@ -276,6 +276,8 @@ impl HttpServer {
 pub struct HttpOptions {
     pub addr: String,
 
+    /// Request timeout; zero disables it. Frontend raises a nonzero timeout to at least
+    /// the largest active synchronous Prom or shared table batch flush interval plus one second.
     #[serde(with = "humantime_serde")]
     pub timeout: Duration,
 

src/standalone/Cargo.toml

@@ -46,4 +46,5 @@ tokio.workspace = true
 url.workspace = true
 
 [dev-dependencies]
+temp-env.workspace = true
 toml.workspace = true

src/standalone/src/options.rs

@@ -133,6 +133,7 @@ impl Configurable for StandaloneOptions {
             "heartbeat_env_vars",
             "wal.broker_endpoints",
             "event_recorder.event_types",
+            "experimental_pending_rows_batcher.protocols",
         ])
     }
 }
@@ -214,6 +215,28 @@ mod tests {
 
     use crate::options::*;
 
+    #[test]
+    fn test_batcher_protocols_from_env() {
+        temp_env::with_vars(
+            [(
+                "STANDALONE_BATCHER_TEST__EXPERIMENTAL_PENDING_ROWS_BATCHER__PROTOCOLS",
+                Some("influxdb,http_sql"),
+            )],
+            || {
+                let options =
+                    StandaloneOptions::load_layered_options(None, "STANDALONE_BATCHER_TEST")
+                        .unwrap();
+                assert_eq!(
+                    options.experimental_pending_rows_batcher.protocols,
+                    vec![
+                        servers::http::BatchingProtocol::Influxdb,
+                        servers::http::BatchingProtocol::HttpSql
+                    ]
+                );
+            },
+        );
+    }
+
     #[test]
     fn test_protocol_pending_rows_batcher_config() {
         let defaults: StandaloneOptions = toml::from_str("").unwrap();