diff --git a/Cargo.lock b/Cargo.lock
index 338eef58c03..632fdc9b017 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -14736,6 +14736,7 @@ dependencies = [
"flow",
"frontend",
"hostname 0.4.1",
+ "humantime-serde",
"log-store",
"mito2",
"pipeline",
diff --git a/config/config.md b/config/config.md
index dba0d28a0dd..bd9b1944e67 100644
--- a/config/config.md
+++ b/config/config.md
@@ -21,6 +21,8 @@
| `write_bytes_exhausted_policy` | String | `wait` | Policy when write bytes quota is exhausted.
Options: "wait" (default, 10s timeout), "wait()" (e.g., "wait(30s)"), "fail" |
| `init_regions_in_background` | Bool | `false` | Initialize all regions in the background during the startup.
By default, it provides services after all regions have been initialized. |
| `init_regions_parallelism` | Integer | `16` | Parallelism of initializing regions. |
+| `max_concurrent_queries` | Integer | `0` | The maximum concurrent region-read streams. Zero means unlimited.
A single query can consume multiple permits. |
+| `concurrent_query_limiter_timeout` | String | `100ms` | Timeout to acquire a permit when `max_concurrent_queries` is reached.
Only effective when the limiter is enabled. |
| `enable_telemetry` | Bool | `true` | Enable telemetry to collect anonymous usage data. Enabled by default. |
| `runtime` | -- | -- | The runtime options. |
| `runtime.global_rt_size` | Integer | Auto | The number of threads to execute the runtime for global read operations.
Defaults to max(num_cpus, 2). |
diff --git a/config/standalone.example.toml b/config/standalone.example.toml
index 46471332fc8..9f94888d2ae 100644
--- a/config/standalone.example.toml
+++ b/config/standalone.example.toml
@@ -47,6 +47,14 @@ init_regions_in_background = false
## Parallelism of initializing regions.
init_regions_parallelism = 16
+## The maximum concurrent region-read streams. Zero means unlimited.
+## A single query can consume multiple permits.
+max_concurrent_queries = 0
+
+## Timeout to acquire a permit when `max_concurrent_queries` is reached.
+## Only effective when the limiter is enabled.
+concurrent_query_limiter_timeout = "100ms"
+
## Enable telemetry to collect anonymous usage data. Enabled by default.
#+ enable_telemetry = true
diff --git a/src/cmd/src/standalone.rs b/src/cmd/src/standalone.rs
index f96f5bef57d..861c7ee2ff3 100644
--- a/src/cmd/src/standalone.rs
+++ b/src/cmd/src/standalone.rs
@@ -1157,6 +1157,8 @@ mod tests {
let mut file = create_named_temp_file();
let toml_str = r#"
enable_memory_catalog = true
+ max_concurrent_queries = 8
+ concurrent_query_limiter_timeout = "250ms"
[wal]
provider = "raft_engine"
@@ -1215,6 +1217,11 @@ mod tests {
.component;
let fe_opts = options.frontend_options();
let dn_opts = options.datanode_options();
+ assert_eq!(dn_opts.max_concurrent_queries, 8);
+ assert_eq!(
+ dn_opts.concurrent_query_limiter_timeout,
+ Duration::from_millis(250)
+ );
let logging_opts = options.logging;
assert_eq!("127.0.0.1:4000".to_string(), fe_opts.http.addr);
assert_eq!(Duration::from_secs(33), fe_opts.http.timeout);
diff --git a/src/standalone/Cargo.toml b/src/standalone/Cargo.toml
index d89366d2e7e..60bd8f7991c 100644
--- a/src/standalone/Cargo.toml
+++ b/src/standalone/Cargo.toml
@@ -33,6 +33,7 @@ file-engine.workspace = true
flow.workspace = true
frontend.workspace = true
hostname.workspace = true
+humantime-serde.workspace = true
log-store.workspace = true
mito2.workspace = true
pipeline.workspace = true
diff --git a/src/standalone/src/options.rs b/src/standalone/src/options.rs
index ea0b85e3efb..afae37c1668 100644
--- a/src/standalone/src/options.rs
+++ b/src/standalone/src/options.rs
@@ -12,6 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
+use std::time::Duration;
+
use common_base::readable_size::ReadableSize;
use common_config::{Configurable, KvBackendConfig};
use common_event_recorder::EventRecorderOptions;
@@ -75,6 +77,13 @@ pub struct StandaloneOptions {
pub tracing: TracingOptions,
pub init_regions_in_background: bool,
pub init_regions_parallelism: usize,
+ /// Maximum concurrent region-read streams. Zero means unlimited.
+ /// A single query can consume multiple permits.
+ pub max_concurrent_queries: usize,
+ /// Timeout to acquire a permit when `max_concurrent_queries` is reached.
+ /// Only effective when the limiter is enabled.
+ #[serde(with = "humantime_serde")]
+ pub concurrent_query_limiter_timeout: Duration,
pub slow_query: SlowQueryOptions,
pub query: QueryOptions,
pub memory: MemoryOptions,
@@ -120,6 +129,8 @@ impl Default for StandaloneOptions {
tracing: TracingOptions::default(),
init_regions_in_background: false,
init_regions_parallelism: 16,
+ max_concurrent_queries: 0,
+ concurrent_query_limiter_timeout: Duration::from_millis(100),
slow_query: SlowQueryOptions::default(),
query: QueryOptions::default(),
memory: MemoryOptions::default(),
@@ -199,6 +210,8 @@ impl StandaloneOptions {
grpc: cloned_opts.grpc,
init_regions_in_background: cloned_opts.init_regions_in_background,
init_regions_parallelism: cloned_opts.init_regions_parallelism,
+ max_concurrent_queries: cloned_opts.max_concurrent_queries,
+ concurrent_query_limiter_timeout: cloned_opts.concurrent_query_limiter_timeout,
query: cloned_opts.query,
heartbeat_env_vars: cloned_opts.heartbeat_env_vars,
..Default::default()
@@ -238,6 +251,74 @@ mod tests {
);
}
+ #[test]
+ fn test_concurrent_query_limiter_options() {
+ let defaults: StandaloneOptions = toml::from_str("").unwrap();
+ let datanode_defaults = DatanodeOptions::default();
+ assert_eq!(defaults.max_concurrent_queries, 0);
+ assert_eq!(
+ defaults.concurrent_query_limiter_timeout,
+ Duration::from_millis(100)
+ );
+ assert_eq!(
+ defaults.max_concurrent_queries,
+ datanode_defaults.max_concurrent_queries
+ );
+ assert_eq!(
+ defaults.concurrent_query_limiter_timeout,
+ datanode_defaults.concurrent_query_limiter_timeout
+ );
+
+ let options: StandaloneOptions = toml::from_str(
+ r#"max_concurrent_queries = 8
+concurrent_query_limiter_timeout = "250ms""#,
+ )
+ .unwrap();
+ let restored: StandaloneOptions =
+ toml::from_str(&toml::to_string(&options).unwrap()).unwrap();
+ for options in [options, restored] {
+ assert_eq!(options.max_concurrent_queries, 8);
+ assert_eq!(
+ options.concurrent_query_limiter_timeout,
+ Duration::from_millis(250)
+ );
+ let datanode = options.datanode_options();
+ assert_eq!(datanode.max_concurrent_queries, 8);
+ assert_eq!(
+ datanode.concurrent_query_limiter_timeout,
+ Duration::from_millis(250)
+ );
+ }
+ assert!(
+ toml::from_str::("concurrent_query_limiter_timeout = 'invalid'")
+ .is_err()
+ );
+ }
+
+ #[test]
+ fn test_concurrent_query_limiter_options_from_env() {
+ temp_env::with_vars(
+ [
+ ("STANDALONE_LIMITER_TEST__MAX_CONCURRENT_QUERIES", Some("4")),
+ (
+ "STANDALONE_LIMITER_TEST__CONCURRENT_QUERY_LIMITER_TIMEOUT",
+ Some("2s"),
+ ),
+ ],
+ || {
+ let options =
+ StandaloneOptions::load_layered_options(None, "STANDALONE_LIMITER_TEST")
+ .unwrap();
+ let datanode = options.datanode_options();
+ assert_eq!(datanode.max_concurrent_queries, 4);
+ assert_eq!(
+ datanode.concurrent_query_limiter_timeout,
+ Duration::from_secs(2)
+ );
+ },
+ );
+ }
+
#[test]
fn test_pipeline_host_access_config_forwarding() {
let defaults: StandaloneOptions = toml::from_str("").unwrap();
diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs
index 2f5f38199a3..f95a3396e83 100644
--- a/tests-integration/tests/http.rs
+++ b/tests-integration/tests/http.rs
@@ -2835,6 +2835,8 @@ max_in_flight_write_bytes = "0KiB"
write_bytes_exhausted_policy = "wait"
init_regions_in_background = false
init_regions_parallelism = 16
+max_concurrent_queries = 0
+concurrent_query_limiter_timeout = "100ms"
heartbeat_env_vars = []
[http]