From cbe471c26fdb99f02c04bc5d65359af2557970b1 Mon Sep 17 00:00:00 2001 From: WenyXu Date: Wed, 16 Sep 2026 09:53:20 +0000 Subject: [PATCH] fix: load batching protocols from environment and document frontend wiring Signed-off-by: WenyXu --- Cargo.lock | 2 ++ Cargo.toml | 1 + config/config.md | 4 ++-- config/frontend.example.toml | 5 +++-- config/standalone.example.toml | 5 +++-- src/frontend/AGENTS.md | 6 ++++++ src/frontend/Cargo.toml | 1 + src/frontend/src/frontend.rs | 22 ++++++++++++++++++++++ src/servers/src/http.rs | 2 ++ src/standalone/Cargo.toml | 1 + src/standalone/src/options.rs | 23 +++++++++++++++++++++++ 11 files changed, 66 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 61005e91848..09c045e245c 100644 --- a/Cargo.lock +++ b/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", diff --git a/Cargo.toml b/Cargo.toml index 730901925a9..7cd91aa0b58 100644 --- a/Cargo.toml +++ b/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" diff --git a/config/config.md b/config/config.md index f0699a79a9c..6325bb73e54 100644 --- a/config/config.md +++ b/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.
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. | +| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.
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`. | | `http.body_limit` | String | `64MB` | HTTP request body limit.
The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.
Set to 0 to disable limit. | | `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default
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.
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.
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. | +| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.
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`. | | `http.body_limit` | String | `64MB` | HTTP request body limit.
The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.
Set to 0 to disable limit. | | `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default
This allows browser to access http APIs without CORS restrictions | | `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. | diff --git a/config/frontend.example.toml b/config/frontend.example.toml index b77df0c035d..4f70c86a040 100644 --- a/config/frontend.example.toml +++ b/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`. diff --git a/config/standalone.example.toml b/config/standalone.example.toml index b7a7aef5a64..4c533a13611 100644 --- a/config/standalone.example.toml +++ b/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`. diff --git a/src/frontend/AGENTS.md b/src/frontend/AGENTS.md index b5c0871fc11..9740d160164 100644 --- a/src/frontend/AGENTS.md +++ b/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. diff --git a/src/frontend/Cargo.toml b/src/frontend/Cargo.toml index 677dbbbe55f..6ebb5979cd7 100644 --- a/src/frontend/Cargo.toml +++ b/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 diff --git a/src/frontend/src/frontend.rs b/src/frontend/src/frontend.rs index 933adcaefa1..b839493b127 100644 --- a/src/frontend/src/frontend.rs +++ b/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 = Pin> + 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(); diff --git a/src/servers/src/http.rs b/src/servers/src/http.rs index 8681590c552..27411a5f8a6 100644 --- a/src/servers/src/http.rs +++ b/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, diff --git a/src/standalone/Cargo.toml b/src/standalone/Cargo.toml index 3dcb1d5f108..a36983203ee 100644 --- a/src/standalone/Cargo.toml +++ b/src/standalone/Cargo.toml @@ -46,4 +46,5 @@ tokio.workspace = true url.workspace = true [dev-dependencies] +temp-env.workspace = true toml.workspace = true diff --git a/src/standalone/src/options.rs b/src/standalone/src/options.rs index d7caf097085..c6ed74c6f74 100644 --- a/src/standalone/src/options.rs +++ b/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();