diff --git a/config/config.md b/config/config.md
index 16384b406f4..d736c24d848 100644
--- a/config/config.md
+++ b/config/config.md
@@ -34,7 +34,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 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 `pending_rows_batcher.pending_rows_flush_interval`. |
+| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.
When synchronous Prometheus, OTLP metrics, or ordinary-table batching is enabled, a nonzero timeout is
raised to at least the largest active flush interval plus 1 second. The intervals come from
`pending_rows_batcher.logical_table.pending_rows_flush_interval` and `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. |
@@ -75,13 +75,20 @@
| `influxdb` | -- | -- | InfluxDB protocol options. |
| `influxdb.enable` | Bool | `true` | Whether to enable InfluxDB protocol in HTTP API. |
| `influxdb.default_merge_mode` | String | `last_non_null` | Default merge mode for tables automatically created by InfluxDB protocol.
Available values: "last_non_null", "last_row". |
-| `pending_rows_batcher` | -- | -- | Shared experimental ordinary-table batching for opted-in ingestion protocols.
Legacy Prometheus batching settings under prom_store remain supported.
HTTP write protocols sharing this batcher. Omitted or empty disables all entrances.
Supported: influxdb, opentsdb, otlp, logs, loki, splunk, elasticsearch, http_sql, prom.
Prom uses ordinary-table batching without metric engine, otherwise its dedicated batcher.
Effective shared Prom settings take precedence; existing prom_store settings remain compatible. |
+| `pending_rows_batcher` | -- | -- | Ordinary-table batching for opted-in HTTP ingestion protocols.
PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge
queue admission without waiting for storage; later failures cannot be returned to the client.
Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.
OTLP logs, traces and ordinary metrics use this batcher. |
| `pending_rows_batcher.pending_rows_flush_interval` | String | `0s` | Flush interval measured from the first pending submission. Zero disables batching. |
| `pending_rows_batcher.max_batch_rows` | Integer | `100000` | Flush after a complete submission reaches this row threshold. |
| `pending_rows_batcher.max_concurrent_flushes` | Integer | `256` | Maximum concurrent flushes shared by the frontend batcher. |
-| `pending_rows_batcher.worker_channel_capacity` | Integer | `65526` | Maximum queued submissions per table worker. |
+| `pending_rows_batcher.worker_channel_capacity` | Integer | `65536` | Maximum queued submissions per table worker. |
| `pending_rows_batcher.max_inflight_requests` | Integer | `3000` | Maximum admitted original requests awaiting completion. |
| `pending_rows_batcher.flow_notification_queue_capacity` | Integer | `1024` | Maximum number of queued table Flow notifications. |
+| `pending_rows_batcher.logical_table` | -- | -- | Metric-engine logical-table batching for Prom remote write and non-legacy OTLP metrics.
Requires prom_store.with_metric_engine. Logs, traces and legacy metrics are not eligible.
Enable independently with protocols and a nonzero flush interval.
Omitted fields use independent defaults, not parent settings.
Empty protocols or a zero interval disables logical batching without fallback.
Omitting this entire section preserves legacy Prom batching; it does not enable OTLP batching. |
+| `pending_rows_batcher.logical_table.pending_rows_flush_interval` | String | `0s` | Flush interval measured from the first pending submission. Zero disables batching. |
+| `pending_rows_batcher.logical_table.max_batch_rows` | Integer | `100000` | Flush after a complete submission reaches this row threshold. |
+| `pending_rows_batcher.logical_table.max_concurrent_flushes` | Integer | `256` | Maximum concurrent flushes shared by Prom and OTLP metrics. |
+| `pending_rows_batcher.logical_table.worker_channel_capacity` | Integer | `65536` | Maximum queued submissions per physical-table worker. |
+| `pending_rows_batcher.logical_table.max_inflight_requests` | Integer | `3000` | Maximum admitted original requests awaiting completion. |
+| `pending_rows_batcher.logical_table.flow_notification_queue_capacity` | Integer | `1024` | Maximum number of queued logical-table Flow notifications. |
| `jaeger` | -- | -- | Jaeger protocol options. |
| `jaeger.enable` | Bool | `true` | Whether to enable Jaeger protocol in HTTP API. |
| `otlp` | -- | -- | OpenTelemetry protocol options. |
@@ -94,12 +101,6 @@
| `prom_store.with_metric_engine` | Bool | `true` | Whether to store the data from Prometheus remote write in metric engine. |
| `prom_store.prom_validation_mode` | String | `strict` | Whether to enable validation for Prometheus remote write requests.
Available options:
- strict: deny invalid UTF-8 strings (default).
- lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).
- unchecked: do not valid strings. |
| `prom_store.experimental_enable_prometheus_native_histogram` | Bool | `false` | Experimental: enable Prometheus remote write v2 native histogram ingestion. |
-| `prom_store.pending_rows_flush_interval` | String | `0s` | Interval to flush pending rows batcher.
Set to "0s" to disable batching mode in Prometheus Remote Write endpoint |
-| `prom_store.max_batch_rows` | Integer | `100000` | Max rows per pending batch before triggering a flush. |
-| `prom_store.max_concurrent_flushes` | Integer | `256` | Max number of concurrent batch flushes. |
-| `prom_store.worker_channel_capacity` | Integer | `65526` | Capacity of the pending batch worker channel. |
-| `prom_store.max_inflight_requests` | Integer | `3000` | Max inflight write requests before backpressure. |
-| `prom_store.flow_notification_queue_capacity` | Integer | `1024` | Maximum number of logical-table flow notifications waiting in the shared queue. |
| `wal` | -- | -- | The WAL options. |
| `wal.provider` | String | `raft_engine` | The provider of the WAL.
- `raft_engine`: the wal is stored in the local file system by raft-engine.
- `kafka`: it's remote wal that data is stored in Kafka.
- `experimental_object_store`: the wal is stored as objects in an object store.
**Notes: experimental and not supported yet.** |
| `wal.dir` | String | Unset | The directory to store the WAL files.
**It's only used when the provider is `raft_engine`**. |
@@ -289,7 +290,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 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 `pending_rows_batcher.pending_rows_flush_interval`. |
+| `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.
When synchronous Prometheus, OTLP metrics, or ordinary-table batching is enabled, a nonzero timeout is
raised to at least the largest active flush interval plus 1 second. The intervals come from
`pending_rows_batcher.logical_table.pending_rows_flush_interval` and `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. |
@@ -342,13 +343,20 @@
| `influxdb` | -- | -- | InfluxDB protocol options. |
| `influxdb.enable` | Bool | `true` | Whether to enable InfluxDB protocol in HTTP API. |
| `influxdb.default_merge_mode` | String | `last_non_null` | Default merge mode for tables automatically created by InfluxDB protocol.
Available values: "last_non_null", "last_row". |
-| `pending_rows_batcher` | -- | -- | Shared experimental ordinary-table batching for opted-in ingestion protocols.
Legacy Prometheus batching settings under prom_store remain supported.
HTTP write protocols sharing this batcher. Omitted or empty disables all entrances.
Supported: influxdb, opentsdb, otlp, logs, loki, splunk, elasticsearch, http_sql, prom.
Prom uses ordinary-table batching without metric engine, otherwise its dedicated batcher.
Effective shared Prom settings take precedence; existing prom_store settings remain compatible. |
+| `pending_rows_batcher` | -- | -- | Ordinary-table batching for opted-in HTTP ingestion protocols.
PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge
queue admission without waiting for storage; later failures cannot be returned to the client.
Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.
OTLP logs, traces and ordinary metrics use this batcher. |
| `pending_rows_batcher.pending_rows_flush_interval` | String | `0s` | Flush interval measured from the first pending submission. Zero disables batching. |
| `pending_rows_batcher.max_batch_rows` | Integer | `100000` | Flush after a complete submission reaches this row threshold. |
| `pending_rows_batcher.max_concurrent_flushes` | Integer | `256` | Maximum concurrent flushes shared by the frontend batcher. |
-| `pending_rows_batcher.worker_channel_capacity` | Integer | `65526` | Maximum queued submissions per table worker. |
+| `pending_rows_batcher.worker_channel_capacity` | Integer | `65536` | Maximum queued submissions per table worker. |
| `pending_rows_batcher.max_inflight_requests` | Integer | `3000` | Maximum admitted original requests awaiting completion. |
| `pending_rows_batcher.flow_notification_queue_capacity` | Integer | `1024` | Maximum number of queued table Flow notifications. |
+| `pending_rows_batcher.logical_table` | -- | -- | Metric-engine logical-table batching for Prom remote write and non-legacy OTLP metrics.
Requires prom_store.with_metric_engine. Logs, traces and legacy metrics are not eligible.
Enable independently with protocols and a nonzero flush interval.
Omitted fields use independent defaults, not parent settings.
Empty protocols or a zero interval disables logical batching without fallback.
Omitting this entire section preserves legacy Prom batching; it does not enable OTLP batching. |
+| `pending_rows_batcher.logical_table.pending_rows_flush_interval` | String | `0s` | Flush interval measured from the first pending submission. Zero disables batching. |
+| `pending_rows_batcher.logical_table.max_batch_rows` | Integer | `100000` | Flush after a complete submission reaches this row threshold. |
+| `pending_rows_batcher.logical_table.max_concurrent_flushes` | Integer | `256` | Maximum concurrent flushes shared by Prom and OTLP metrics. |
+| `pending_rows_batcher.logical_table.worker_channel_capacity` | Integer | `65536` | Maximum queued submissions per physical-table worker. |
+| `pending_rows_batcher.logical_table.max_inflight_requests` | Integer | `3000` | Maximum admitted original requests awaiting completion. |
+| `pending_rows_batcher.logical_table.flow_notification_queue_capacity` | Integer | `1024` | Maximum number of queued logical-table Flow notifications. |
| `jaeger` | -- | -- | Jaeger protocol options. |
| `jaeger.enable` | Bool | `true` | Whether to enable Jaeger protocol in HTTP API. |
| `otlp` | -- | -- | OpenTelemetry protocol options. |
@@ -361,12 +369,6 @@
| `prom_store.with_metric_engine` | Bool | `true` | Whether to store the data from Prometheus remote write in metric engine. |
| `prom_store.prom_validation_mode` | String | `strict` | Whether to enable validation for Prometheus remote write requests.
Available options:
- strict: deny invalid UTF-8 strings (default).
- lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).
- unchecked: do not valid strings. |
| `prom_store.experimental_enable_prometheus_native_histogram` | Bool | `false` | Experimental: enable Prometheus remote write v2 native histogram ingestion. |
-| `prom_store.pending_rows_flush_interval` | String | `0s` | Interval to flush pending rows batcher.
Set to "0s" to disable batching mode in Prometheus Remote Write endpoint |
-| `prom_store.max_batch_rows` | Integer | `100000` | Max rows per pending batch before triggering a flush. |
-| `prom_store.max_concurrent_flushes` | Integer | `256` | Max number of concurrent batch flushes. |
-| `prom_store.worker_channel_capacity` | Integer | `65526` | Capacity of the pending batch worker channel. |
-| `prom_store.max_inflight_requests` | Integer | `3000` | Max inflight write requests before backpressure. |
-| `prom_store.flow_notification_queue_capacity` | Integer | `1024` | Maximum number of logical-table flow notifications waiting in the shared queue. |
| `meta_client` | -- | -- | The metasrv client options. |
| `meta_client.metasrv_addrs` | Array | -- | The addresses of the metasrv. |
| `meta_client.timeout` | String | `3s` | Operation timeout. |
diff --git a/config/frontend.example.toml b/config/frontend.example.toml
index 043da57db74..275db3105c5 100644
--- a/config/frontend.example.toml
+++ b/config/frontend.example.toml
@@ -56,9 +56,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 synchronous Prometheus or shared table batching is enabled, a nonzero timeout is
+## When synchronous Prometheus, OTLP metrics, or ordinary-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 `pending_rows_batcher.pending_rows_flush_interval`.
+## `pending_rows_batcher.logical_table.pending_rows_flush_interval` and `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`.
@@ -231,12 +231,11 @@ enable = true
## Available values: "last_non_null", "last_row".
default_merge_mode = "last_non_null"
-## Shared experimental ordinary-table batching for opted-in ingestion protocols.
-## Legacy Prometheus batching settings under prom_store remain supported.
-## HTTP write protocols sharing this batcher. Omitted or empty disables all entrances.
-## Supported: influxdb, opentsdb, otlp, logs, loki, splunk, elasticsearch, http_sql, prom.
-## Prom uses ordinary-table batching without metric engine, otherwise its dedicated batcher.
-## Effective shared Prom settings take precedence; existing prom_store settings remain compatible.
+## Ordinary-table batching for opted-in HTTP ingestion protocols.
+## PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge
+## queue admission without waiting for storage; later failures cannot be returned to the client.
+## Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.
+## OTLP logs, traces and ordinary metrics use this batcher.
[pending_rows_batcher]
# protocols = [
# "influxdb",
@@ -256,12 +255,33 @@ max_batch_rows = 100000
## Maximum concurrent flushes shared by the frontend batcher.
max_concurrent_flushes = 256
## Maximum queued submissions per table worker.
-worker_channel_capacity = 65526
+worker_channel_capacity = 65536
## Maximum admitted original requests awaiting completion.
max_inflight_requests = 3000
## Maximum number of queued table Flow notifications.
flow_notification_queue_capacity = 1024
+## Metric-engine logical-table batching for Prom remote write and non-legacy OTLP metrics.
+## Requires prom_store.with_metric_engine. Logs, traces and legacy metrics are not eligible.
+## Enable independently with protocols and a nonzero flush interval.
+## Omitted fields use independent defaults, not parent settings.
+## Empty protocols or a zero interval disables logical batching without fallback.
+## Omitting this entire section preserves legacy Prom batching; it does not enable OTLP batching.
+[pending_rows_batcher.logical_table]
+# protocols = ["prom", "otlp"]
+## Flush interval measured from the first pending submission. Zero disables batching.
+pending_rows_flush_interval = "0s"
+## Flush after a complete submission reaches this row threshold.
+max_batch_rows = 100000
+## Maximum concurrent flushes shared by Prom and OTLP metrics.
+max_concurrent_flushes = 256
+## Maximum queued submissions per physical-table worker.
+worker_channel_capacity = 65536
+## Maximum admitted original requests awaiting completion.
+max_inflight_requests = 3000
+## Maximum number of queued logical-table Flow notifications.
+flow_notification_queue_capacity = 1024
+
## Jaeger protocol options.
[jaeger]
## Whether to enable Jaeger protocol in HTTP API.
@@ -293,19 +313,7 @@ with_metric_engine = true
prom_validation_mode = "strict"
## Experimental: enable Prometheus remote write v2 native histogram ingestion.
experimental_enable_prometheus_native_histogram = false
-## Interval to flush pending rows batcher.
-## Set to "0s" to disable batching mode in Prometheus Remote Write endpoint
-#+pending_rows_flush_interval = "0s"
-## Max rows per pending batch before triggering a flush.
-#+max_batch_rows = 100000
-## Max number of concurrent batch flushes.
-#+max_concurrent_flushes = 256
-## Capacity of the pending batch worker channel.
-#+worker_channel_capacity = 65526
-## Max inflight write requests before backpressure.
-#+max_inflight_requests = 3000
-## Maximum number of logical-table flow notifications waiting in the shared queue.
-#+flow_notification_queue_capacity = 1024
+
## The metasrv client options.
[meta_client]
diff --git a/config/standalone.example.toml b/config/standalone.example.toml
index e59e7ba1b4c..257f48eda2c 100644
--- a/config/standalone.example.toml
+++ b/config/standalone.example.toml
@@ -81,9 +81,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 synchronous Prometheus or shared table batching is enabled, a nonzero timeout is
+## When synchronous Prometheus, OTLP metrics, or ordinary-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 `pending_rows_batcher.pending_rows_flush_interval`.
+## `pending_rows_batcher.logical_table.pending_rows_flush_interval` and `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`.
@@ -210,12 +210,11 @@ enable = true
## Available values: "last_non_null", "last_row".
default_merge_mode = "last_non_null"
-## Shared experimental ordinary-table batching for opted-in ingestion protocols.
-## Legacy Prometheus batching settings under prom_store remain supported.
-## HTTP write protocols sharing this batcher. Omitted or empty disables all entrances.
-## Supported: influxdb, opentsdb, otlp, logs, loki, splunk, elasticsearch, http_sql, prom.
-## Prom uses ordinary-table batching without metric engine, otherwise its dedicated batcher.
-## Effective shared Prom settings take precedence; existing prom_store settings remain compatible.
+## Ordinary-table batching for opted-in HTTP ingestion protocols.
+## PENDING_ROWS_BATCH_SYNC defaults to true for both batchers. Set it to false to acknowledge
+## queue admission without waiting for storage; later failures cannot be returned to the client.
+## Omitted or empty protocols disables batching. Prom without metric engine uses this batcher.
+## OTLP logs, traces and ordinary metrics use this batcher.
[pending_rows_batcher]
# protocols = [
# "influxdb",
@@ -235,12 +234,33 @@ max_batch_rows = 100000
## Maximum concurrent flushes shared by the frontend batcher.
max_concurrent_flushes = 256
## Maximum queued submissions per table worker.
-worker_channel_capacity = 65526
+worker_channel_capacity = 65536
## Maximum admitted original requests awaiting completion.
max_inflight_requests = 3000
## Maximum number of queued table Flow notifications.
flow_notification_queue_capacity = 1024
+## Metric-engine logical-table batching for Prom remote write and non-legacy OTLP metrics.
+## Requires prom_store.with_metric_engine. Logs, traces and legacy metrics are not eligible.
+## Enable independently with protocols and a nonzero flush interval.
+## Omitted fields use independent defaults, not parent settings.
+## Empty protocols or a zero interval disables logical batching without fallback.
+## Omitting this entire section preserves legacy Prom batching; it does not enable OTLP batching.
+[pending_rows_batcher.logical_table]
+# protocols = ["prom", "otlp"]
+## Flush interval measured from the first pending submission. Zero disables batching.
+pending_rows_flush_interval = "0s"
+## Flush after a complete submission reaches this row threshold.
+max_batch_rows = 100000
+## Maximum concurrent flushes shared by Prom and OTLP metrics.
+max_concurrent_flushes = 256
+## Maximum queued submissions per physical-table worker.
+worker_channel_capacity = 65536
+## Maximum admitted original requests awaiting completion.
+max_inflight_requests = 3000
+## Maximum number of queued logical-table Flow notifications.
+flow_notification_queue_capacity = 1024
+
## Jaeger protocol options.
[jaeger]
## Whether to enable Jaeger protocol in HTTP API.
@@ -272,19 +292,7 @@ with_metric_engine = true
prom_validation_mode = "strict"
## Experimental: enable Prometheus remote write v2 native histogram ingestion.
experimental_enable_prometheus_native_histogram = false
-## Interval to flush pending rows batcher.
-## Set to "0s" to disable batching mode in Prometheus Remote Write endpoint
-#+pending_rows_flush_interval = "0s"
-## Max rows per pending batch before triggering a flush.
-#+max_batch_rows = 100000
-## Max number of concurrent batch flushes.
-#+max_concurrent_flushes = 256
-## Capacity of the pending batch worker channel.
-#+worker_channel_capacity = 65526
-## Max inflight write requests before backpressure.
-#+max_inflight_requests = 3000
-## Maximum number of logical-table flow notifications waiting in the shared queue.
-#+flow_notification_queue_capacity = 1024
+
## The WAL options.
[wal]
diff --git a/src/cmd/src/bin/query_perf_fixture/case.rs b/src/cmd/src/bin/query_perf_fixture/case.rs
index 0f5239a6b51..271c3928e35 100644
--- a/src/cmd/src/bin/query_perf_fixture/case.rs
+++ b/src/cmd/src/bin/query_perf_fixture/case.rs
@@ -185,7 +185,7 @@ pub(super) fn default_max_concurrent_flushes() -> u64 {
256
}
pub(super) fn default_worker_channel_capacity() -> u64 {
- 65526
+ 65_536
}
pub(super) fn default_max_inflight_requests() -> u64 {
3000
diff --git a/src/cmd/tests/load_config_test.rs b/src/cmd/tests/load_config_test.rs
index 67e2e573f38..1c4a83e1397 100644
--- a/src/cmd/tests/load_config_test.rs
+++ b/src/cmd/tests/load_config_test.rs
@@ -30,6 +30,7 @@ use datanode::config::{DatanodeOptions, RegionEngineConfig, StorageConfig};
use file_engine::config::EngineConfig as FileEngineConfig;
use flow::FlownodeOptions;
use frontend::frontend::FrontendOptions;
+use frontend::service_config::PendingRowsBatcherOptions;
use meta_client::MetaClientOptions;
use meta_srv::metasrv::MetasrvOptions;
use meta_srv::selector::SelectorType;
@@ -206,6 +207,10 @@ fn test_load_frontend_example_config() {
);
let expected = GreptimeOptions:: {
component: FrontendOptions {
+ pending_rows_batcher: PendingRowsBatcherOptions {
+ logical_table: Some(Default::default()),
+ ..Default::default()
+ },
default_timezone: Some("UTC".to_string()),
default_column_prefix: Some("greptime".to_string()),
auto_create_table: true,
@@ -389,6 +394,10 @@ fn test_load_standalone_example_config() {
);
let expected = GreptimeOptions:: {
component: StandaloneOptions {
+ pending_rows_batcher: PendingRowsBatcherOptions {
+ logical_table: Some(Default::default()),
+ ..Default::default()
+ },
default_timezone: Some("UTC".to_string()),
default_column_prefix: Some("greptime".to_string()),
auto_create_table: true,
diff --git a/src/frontend/AGENTS.md b/src/frontend/AGENTS.md
index d552d73cf39..49a4b8fcd0f 100644
--- a/src/frontend/AGENTS.md
+++ b/src/frontend/AGENTS.md
@@ -51,6 +51,11 @@ remote datanodes via `operator`/`client`.
- Internal gRPC listeners mark requests with `Channel::Internal` in middleware
(`server.rs`), including requests handled by Enterprise Flight wrappers.
+- **Logical-table batching** (`instance/logical_batcher.rs`): `Services` initializes
+ one shared batcher for opted-in HTTP Prom and nonlegacy OTLP metric-engine
+ writes. OTLP checks operator eligibility and falls back for incompatible tables.
+ The schema adapter holds a weak instance reference to avoid an ownership cycle.
+
- **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
diff --git a/src/frontend/src/frontend.rs b/src/frontend/src/frontend.rs
index 86ba2cb9bfb..72bf322b1f1 100644
--- a/src/frontend/src/frontend.rs
+++ b/src/frontend/src/frontend.rs
@@ -67,7 +67,7 @@ pub struct FrontendOptions {
pub postgres: PostgresOptions,
pub opentsdb: OpentsdbOptions,
pub influxdb: InfluxdbOptions,
- /// Shared experimental ordinary-table batching; independent of Prom batching.
+ /// Ordinary-table batching with independent logical-table controls.
pub pending_rows_batcher: PendingRowsBatcherOptions,
pub prom_store: PromStoreOptions,
pub jaeger: JaegerOptions,
@@ -131,6 +131,7 @@ impl Configurable for FrontendOptions {
"meta_client.metasrv_addrs",
"event_recorder.event_types",
"pending_rows_batcher.protocols",
+ "pending_rows_batcher.logical_table.protocols",
])
}
}
@@ -225,6 +226,38 @@ mod tests {
type GrpcStream =
Pin> + Send + Sync + 'static>>;
+ #[test]
+ fn test_logical_batcher_protocols_from_env() {
+ temp_env::with_vars(
+ [
+ (
+ "FRONTEND_LOGICAL_TEST__PENDING_ROWS_BATCHER__PROTOCOLS",
+ Some("otlp,influxdb"),
+ ),
+ (
+ "FRONTEND_LOGICAL_TEST__PENDING_ROWS_BATCHER__LOGICAL_TABLE__PROTOCOLS",
+ Some("prom,otlp"),
+ ),
+ ],
+ || {
+ let options =
+ FrontendOptions::load_layered_options(None, "FRONTEND_LOGICAL_TEST").unwrap();
+ assert_eq!(options.pending_rows_batcher.table.protocols.len(), 2);
+ assert_eq!(
+ options
+ .pending_rows_batcher
+ .logical_table
+ .unwrap()
+ .protocols,
+ vec![
+ servers::http::BatchingProtocol::Prom,
+ servers::http::BatchingProtocol::Otlp
+ ]
+ );
+ },
+ );
+ }
+
#[test]
fn test_batcher_protocols_from_env() {
temp_env::with_vars(
@@ -236,7 +269,7 @@ mod tests {
let options =
FrontendOptions::load_layered_options(None, "FRONTEND_BATCHER_TEST").unwrap();
assert_eq!(
- options.pending_rows_batcher.protocols,
+ options.pending_rows_batcher.table.protocols,
vec![
servers::http::BatchingProtocol::Influxdb,
servers::http::BatchingProtocol::HttpSql
@@ -252,6 +285,7 @@ mod tests {
assert!(
!defaults
.pending_rows_batcher
+ .table
.pending_rows_batching_enabled()
);
let options: FrontendOptions = toml::from_str(
@@ -263,9 +297,14 @@ max_batch_rows = 25
"#,
)
.unwrap();
- assert_eq!(options.pending_rows_batcher.max_batch_rows, 25);
- assert_eq!(options.pending_rows_batcher.protocols.len(), 2);
- assert!(options.pending_rows_batcher.pending_rows_batching_enabled());
+ assert_eq!(options.pending_rows_batcher.table.max_batch_rows, 25);
+ assert_eq!(options.pending_rows_batcher.table.protocols.len(), 2);
+ assert!(
+ options
+ .pending_rows_batcher
+ .table
+ .pending_rows_batching_enabled()
+ );
let serialized = toml::to_string(&options).unwrap();
let parsed: FrontendOptions = toml::from_str(&serialized).unwrap();
assert_eq!(options.influxdb, parsed.influxdb);
diff --git a/src/frontend/src/instance.rs b/src/frontend/src/instance.rs
index 22d2afaa7bc..92bc74ca10b 100644
--- a/src/frontend/src/instance.rs
+++ b/src/frontend/src/instance.rs
@@ -21,6 +21,7 @@ mod import_packed;
mod influxdb;
mod jaeger;
mod log_handler;
+mod logical_batcher;
mod logs;
mod opentsdb;
mod otlp;
@@ -31,7 +32,7 @@ mod region_query;
use std::collections::HashSet;
use std::pin::Pin;
use std::sync::atomic::AtomicBool;
-use std::sync::{Arc, atomic};
+use std::sync::{Arc, OnceLock, atomic};
use std::time::{Duration, SystemTime};
use async_stream::stream;
@@ -78,6 +79,7 @@ use query::metrics::OnDone;
use query::parser::{PromQuery, QueryStatement};
use query::query_engine::DescribeResult;
use query::query_engine::options::{QueryOptions, validate_catalog_and_schema};
+use servers::batcher::logical_table::LogicalTablePendingRowsBatcher;
use servers::error::{
self as server_error, AuthSnafu, CommonMetaSnafu, ExecuteQuerySnafu,
OtlpMetricModeIncompatibleSnafu, UnexpectedResultSnafu,
@@ -130,6 +132,7 @@ pub struct Instance {
query_engine: QueryEngineRef,
plugins: Plugins,
inserter: InserterRef,
+ logical_batcher: Arc>>>,
deleter: DeleterRef,
table_metadata_manager: TableMetadataManagerRef,
event_recorder: EventRecorderRef,
diff --git a/src/frontend/src/instance/builder.rs b/src/frontend/src/instance/builder.rs
index 01e9ef034e3..a3cfbcdaffe 100644
--- a/src/frontend/src/instance/builder.rs
+++ b/src/frontend/src/instance/builder.rs
@@ -61,7 +61,7 @@ use crate::heartbeat::frontend_peer_addr;
use crate::instance::Instance;
use crate::instance::entity_graph::EntityGraphProviderImpl;
use crate::instance::region_query::FrontendRegionQueryHandler;
-use crate::service_config::PendingRowsBatcherOptions;
+use crate::service_config::BatcherOptions;
/// The frontend [`Instance`] builder.
pub struct FrontendBuilder {
@@ -237,31 +237,30 @@ impl FrontendBuilder {
};
// The execution-only inserter owns no batchers, avoiding an Arc cycle.
let bulk_inserter = Arc::new(create_inserter());
- let build_batcher =
- |options: &PendingRowsBatcherOptions| -> Option> {
- if !options.pending_rows_batching_enabled()
- || (self.options.prom_store.with_metric_engine
- && options
- .protocols
- .iter()
- .all(|protocol| *protocol == BatchingProtocol::Prom))
- {
- return None;
- }
- TablePendingRowsBatcher::try_new(
- options.pending_rows_flush_interval,
- options.max_batch_rows,
- options.max_concurrent_flushes,
- options.worker_channel_capacity,
- options.max_inflight_requests,
- options.flow_notification_queue_capacity,
- bulk_inserter.clone(),
- )
- .map(|batcher| batcher as Arc)
- };
+ let build_batcher = |options: &BatcherOptions| -> Option> {
+ if !options.pending_rows_batching_enabled()
+ || (self.options.prom_store.with_metric_engine
+ && options
+ .protocols
+ .iter()
+ .all(|protocol| *protocol == BatchingProtocol::Prom))
+ {
+ return None;
+ }
+ TablePendingRowsBatcher::try_new(
+ options.pending_rows_flush_interval,
+ options.max_batch_rows,
+ options.max_concurrent_flushes,
+ options.worker_channel_capacity,
+ options.max_inflight_requests,
+ options.flow_notification_queue_capacity,
+ bulk_inserter.clone(),
+ )
+ .map(|batcher| batcher as Arc)
+ };
let inserter = Arc::new(
create_inserter()
- .with_pending_rows_batcher(build_batcher(&self.options.pending_rows_batcher)),
+ .with_pending_rows_batcher(build_batcher(self.options.table_batcher_options())),
);
let deleter = Arc::new(Deleter::new(
self.catalog_manager.clone(),
@@ -386,6 +385,7 @@ impl FrontendBuilder {
admin_event_recorder.install(&event_recorder);
Ok(Instance {
+ logical_batcher: Default::default(),
frontend_peer_addr,
experimental_metric_export: self.options.experimental_metric_export,
catalog_manager: self.catalog_manager,
diff --git a/src/frontend/src/instance/logical_batcher.rs b/src/frontend/src/instance/logical_batcher.rs
new file mode 100644
index 00000000000..28ed19b3e14
--- /dev/null
+++ b/src/frontend/src/instance/logical_batcher.rs
@@ -0,0 +1,102 @@
+// 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::sync::{Arc, Weak};
+
+use api::v1::ColumnSchema;
+use async_trait::async_trait;
+use servers::batcher::logical_table::{LogicalTablePendingRowsBatcher, PendingRowsSchemaAlterer};
+use servers::error::{BatcherChannelClosedSnafu, Result};
+use servers::http::BatchingProtocol;
+use session::context::QueryContextRef;
+use snafu::OptionExt;
+
+use crate::frontend::FrontendOptions;
+use crate::instance::Instance;
+
+impl Instance {
+ pub(crate) fn init_logical_batcher(self: &Arc, options: &FrontendOptions) {
+ self.logical_batcher.get_or_init(|| {
+ let options_batcher = options.logical_batcher_options();
+ let enabled = options_batcher
+ .protocols
+ .iter()
+ .any(|protocol| match protocol {
+ BatchingProtocol::Prom => options.prom_store.enable,
+ BatchingProtocol::Otlp => options.otlp.enable,
+ _ => false,
+ });
+ if !options.prom_store.with_metric_engine
+ || !enabled
+ || !options_batcher.pending_rows_batching_enabled()
+ {
+ return None;
+ }
+ LogicalTablePendingRowsBatcher::try_new(
+ self.partition_manager().clone(),
+ self.node_manager().clone(),
+ self.catalog_manager().clone(),
+ self.table_flownode_set_cache().clone(),
+ true,
+ Arc::new(LogicalTables(Arc::downgrade(self))),
+ options_batcher.pending_rows_flush_interval,
+ options_batcher.max_batch_rows,
+ options_batcher.max_concurrent_flushes,
+ options_batcher.worker_channel_capacity,
+ options_batcher.max_inflight_requests,
+ options_batcher.flow_notification_queue_capacity,
+ )
+ });
+ }
+
+ pub(crate) fn logical_batcher(&self) -> Option<&Arc> {
+ self.logical_batcher.get().and_then(Option::as_ref)
+ }
+}
+
+// The instance owns the shared batcher; schema preparation must not keep that
+// owner alive through a reference cycle.
+struct LogicalTables(Weak);
+
+#[async_trait]
+impl PendingRowsSchemaAlterer for LogicalTables {
+ async fn create_tables_if_missing_batch(
+ &self,
+ catalog: &str,
+ schema: &str,
+ tables: &[(&str, &[ColumnSchema])],
+ with_metric_engine: bool,
+ ctx: QueryContextRef,
+ ) -> Result<()> {
+ self.0
+ .upgrade()
+ .context(BatcherChannelClosedSnafu)?
+ .create_tables_if_missing_batch(catalog, schema, tables, with_metric_engine, ctx)
+ .await
+ }
+
+ async fn add_missing_prom_tag_columns_batch(
+ &self,
+ catalog: &str,
+ schema: &str,
+ tables: &[(&str, &[String])],
+ ctx: QueryContextRef,
+ ) -> Result<()> {
+ self.0
+ .upgrade()
+ .context(BatcherChannelClosedSnafu)?
+ .add_missing_prom_tag_columns_batch(catalog, schema, tables, ctx)
+ .await
+ }
+}
diff --git a/src/frontend/src/instance/otlp.rs b/src/frontend/src/instance/otlp.rs
index fa40c7eb59a..7a0f706c58d 100644
--- a/src/frontend/src/instance/otlp.rs
+++ b/src/frontend/src/instance/otlp.rs
@@ -27,10 +27,11 @@ use client::Output;
use common_catalog::consts::{trace_operations_table_name, trace_services_table_name};
use common_error::ext::BoxedError;
use common_query::prelude::GREPTIME_PHYSICAL_TABLE;
+use common_query::{OutputData, OutputMeta};
use common_telemetry::{tracing, warn};
use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
-use operator::insert::{admit_row_insert_batches, admit_write};
+use operator::insert::{Inserter, admit_row_insert_batches, admit_write};
use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest;
use pipeline::{GreptimePipelineParams, PipelineWay};
use servers::error::{self, AuthSnafu, Result as ServerResult};
@@ -165,27 +166,59 @@ impl OpenTelemetryProtocolHandler for Instance {
Arc::new(c)
};
+ let physical_table = ctx
+ .extension(PHYSICAL_TABLE_PARAM)
+ .unwrap_or(GREPTIME_PHYSICAL_TABLE)
+ .to_string();
+ let batcher = self.logical_batcher().filter(|_| {
+ ctx.logical_batching_enabled() && !metric_ctx.is_legacy && metric_ctx.with_metric_engine
+ });
+ let batcher = if batcher.is_some()
+ && self
+ .inserter
+ .can_batch_metric_rows(&requests, &ctx, &physical_table)
+ .await
+ .map_err(BoxedError::new)
+ .context(error::ExecuteGrpcQuerySnafu)?
+ {
+ batcher
+ } else {
+ None
+ };
+
// OTLP tables have one sample field in both the legacy and physical paths.
- let output = if metric_ctx.is_legacy || !metric_ctx.with_metric_engine {
+ let output = if let Some(batcher) = batcher {
+ let (rows, cost) = batcher
+ .submit_with(requests, ctx.clone(), |mut requests| {
+ let ctx = ctx.clone();
+ async move {
+ Inserter::meter_row_inserts(&mut requests, &ctx)
+ .await
+ .map_err(BoxedError::new)
+ .context(error::ExecuteGrpcQuerySnafu)
+ }
+ })
+ .await?;
+ Output::new(
+ OutputData::AffectedRows(rows as usize),
+ OutputMeta::new_with_cost(cost as _),
+ )
+ } else if metric_ctx.is_legacy || !metric_ctx.with_metric_engine {
self.handle_row_inserts(requests, ctx.clone(), false, true)
.await
.map_err(BoxedError::new)
- .context(error::ExecuteGrpcQuerySnafu)
+ .context(error::ExecuteGrpcQuerySnafu)?
} else {
- let physical_table = ctx
- .extension(PHYSICAL_TABLE_PARAM)
- .unwrap_or(GREPTIME_PHYSICAL_TABLE)
- .to_string();
self.handle_metric_row_inserts(requests, ctx.clone(), physical_table)
.await
.map_err(BoxedError::new)
- .context(error::ExecuteGrpcQuerySnafu)
- }?;
+ .context(error::ExecuteGrpcQuerySnafu)?
+ };
outcome.write_cost = output.meta.cost;
- // Derived enrichment, written after the metric data is committed:
- // failing here would make the client retry data the server already
- // accepted, so every failure degrades to a warning instead.
+ // Derived enrichment follows the accepted metric submission, which may
+ // still be queued in asynchronous mode. Failures remain warning-only
+ // to avoid retrying metric data the server already accepted.
if let Some(resource_info) = resource_info {
let written = match self.check_row_insert_permission(
&resource_info,
diff --git a/src/frontend/src/server.rs b/src/frontend/src/server.rs
index 017bba53fba..0a6612ba31c 100644
--- a/src/frontend/src/server.rs
+++ b/src/frontend/src/server.rs
@@ -24,9 +24,7 @@ use common_base::Plugins;
use common_config::Configurable;
use common_telemetry::{info, warn};
use meta_client::MetaClientOptions;
-use servers::batcher::logical_table::{
- LogicalTablePendingRowsBatcher, pending_rows_batch_sync_enabled,
-};
+use servers::batcher::pending_rows_batch_sync_enabled;
use servers::error::Error as ServerError;
use servers::grpc::builder::GrpcServerBuilder;
use servers::grpc::flight::FlightCraftRef;
@@ -74,6 +72,7 @@ where
{
pub fn new(opts: T, instance: Arc, plugins: Plugins) -> Self {
let feopts = opts.clone().into();
+ instance.init_logical_batcher(&feopts);
// Create server request memory limiter for all server protocols
let server_memory_limiter = ServerMemoryLimiter::new(
feopts.max_in_flight_write_bytes.as_bytes(),
@@ -110,7 +109,8 @@ where
request_memory_limiter: ServerMemoryLimiter,
) -> HttpServerBuilder {
let mut builder = HttpServerBuilder::new(effective_http_options(opts))
- .with_batching_protocols(opts.pending_rows_batcher.protocols.clone())
+ .with_batching_protocols(opts.table_batcher_options().protocols.clone())
+ .with_logical_batching_protocols(opts.logical_batcher_options().protocols)
.with_memory_limiter(request_memory_limiter)
.with_sql_handler(self.instance.clone());
@@ -134,24 +134,12 @@ where
let prom_store = effective_prom_store_options(opts);
if prom_store.enable {
- let pending_rows_batcher = if prom_store.with_metric_engine {
- LogicalTablePendingRowsBatcher::try_new(
- self.instance.partition_manager().clone(),
- self.instance.node_manager().clone(),
- self.instance.catalog_manager().clone(),
- self.instance.table_flownode_set_cache().clone(),
- prom_store.with_metric_engine,
- self.instance.clone(),
- prom_store.pending_rows_flush_interval,
- prom_store.max_batch_rows,
- prom_store.max_concurrent_flushes,
- prom_store.worker_channel_capacity,
- prom_store.max_inflight_requests,
- prom_store.flow_notification_queue_capacity,
- )
- } else {
- None
- };
+ let pending_rows_batcher = opts
+ .logical_batcher_options()
+ .protocols
+ .contains(&BatchingProtocol::Prom)
+ .then(|| self.instance.logical_batcher().cloned())
+ .flatten();
builder = builder
.with_prom_handler(
self.instance.clone(),
@@ -439,15 +427,16 @@ where
/// Selected shared controls override legacy Prom batching knobs, not protocol behavior.
fn effective_prom_store_options(opts: &FrontendOptions) -> PromStoreOptions {
let mut prom_store = opts.prom_store.clone();
- let shared = &opts.pending_rows_batcher;
- if shared.protocols.contains(&BatchingProtocol::Prom) && shared.pending_rows_batching_enabled()
- {
+ let shared = opts.logical_batcher_options();
+ if shared.protocols.contains(&BatchingProtocol::Prom) {
prom_store.pending_rows_flush_interval = shared.pending_rows_flush_interval;
prom_store.max_batch_rows = shared.max_batch_rows;
prom_store.max_concurrent_flushes = shared.max_concurrent_flushes;
prom_store.worker_channel_capacity = shared.worker_channel_capacity;
prom_store.max_inflight_requests = shared.max_inflight_requests;
prom_store.flow_notification_queue_capacity = shared.flow_notification_queue_capacity;
+ } else {
+ prom_store.pending_rows_flush_interval = Duration::ZERO;
}
prom_store
}
@@ -459,10 +448,9 @@ fn effective_http_options(opts: &FrontendOptions) -> HttpOptions {
fn effective_http_options_with_sync(opts: &FrontendOptions, batch_sync: bool) -> HttpOptions {
let mut http = opts.http.clone();
let prom_store = effective_prom_store_options(opts);
- let shared = &opts.pending_rows_batcher;
- // Ordinary-table batching always waits for its flush, independently of the
- // dedicated Prom batcher's asynchronous acknowledgement mode.
- let common_enabled = shared.pending_rows_batching_enabled()
+ let shared = opts.table_batcher_options();
+ let common_enabled = batch_sync
+ && shared.pending_rows_batching_enabled()
&& shared.protocols.iter().any(|protocol| {
*protocol != BatchingProtocol::Prom
|| (prom_store.enable && !prom_store.with_metric_engine)
@@ -470,7 +458,19 @@ fn effective_http_options_with_sync(opts: &FrontendOptions, batch_sync: bool) ->
let common_interval = common_enabled.then_some(shared.pending_rows_flush_interval);
let prom_interval = (prom_store.pending_rows_batching_enabled() && batch_sync)
.then_some(prom_store.pending_rows_flush_interval);
- let Some(flush_interval) = common_interval.into_iter().chain(prom_interval).max() else {
+ let logical = opts.logical_batcher_options();
+ let otlp_interval = (batch_sync
+ && opts.otlp.enable
+ && opts.prom_store.with_metric_engine
+ && logical.protocols.contains(&BatchingProtocol::Otlp)
+ && logical.pending_rows_batching_enabled())
+ .then_some(logical.pending_rows_flush_interval);
+ let Some(flush_interval) = common_interval
+ .into_iter()
+ .chain(prom_interval)
+ .chain(otlp_interval)
+ .max()
+ else {
return http;
};
let fallback_timeout = flush_interval.saturating_add(Duration::from_secs(1));
@@ -515,6 +515,81 @@ mod tests {
use crate::instance::builder::FrontendBuilder;
use crate::server::*;
+ #[tokio::test]
+ async fn test_logical_batcher_shared_without_prom_endpoint() {
+ use crate::service_config::BatcherOptions;
+ let mut options = FrontendOptions::default();
+ options.prom_store.enable = false;
+ options.pending_rows_batcher.logical_table = Some(BatcherOptions {
+ protocols: vec![BatchingProtocol::Otlp],
+ pending_rows_flush_interval: Duration::from_millis(5),
+ ..Default::default()
+ });
+ let meta_client = Arc::new(
+ MetaClientBuilder::new(0, Role::Frontend)
+ .enable_procedure()
+ .build(),
+ );
+ let instance = Arc::new(
+ FrontendBuilder::new_test(&options, meta_client)
+ .try_build()
+ .await
+ .unwrap(),
+ );
+ let services = Services::new(options.clone(), instance.clone(), Plugins::default());
+ let batcher = instance.logical_batcher().unwrap().clone();
+ instance.init_logical_batcher(&options);
+ assert!(Arc::ptr_eq(&batcher, instance.logical_batcher().unwrap()));
+ let weak = Arc::downgrade(&instance);
+ drop(services);
+ drop(instance);
+ assert!(
+ weak.upgrade().is_none(),
+ "batcher must not retain its schema owner"
+ );
+ }
+
+ #[test]
+ fn test_logical_batcher_http_timeout_and_prom_disable() {
+ use crate::service_config::pending_rows_batcher::BatcherOptions;
+ let mut opts = FrontendOptions::default();
+ opts.http.timeout = Duration::from_secs(1);
+ opts.prom_store.pending_rows_flush_interval = Duration::from_secs(2);
+ opts.pending_rows_batcher.logical_table = Some(BatcherOptions {
+ protocols: vec![BatchingProtocol::Otlp],
+ pending_rows_flush_interval: Duration::from_secs(5),
+ ..Default::default()
+ });
+ assert!(!effective_prom_store_options(&opts).pending_rows_batching_enabled());
+ // Both logical protocols follow the global acknowledgement policy,
+ // independently of enabling the Prom HTTP endpoint.
+ opts.prom_store.enable = false;
+ assert_eq!(
+ effective_http_options_with_sync(&opts, true).timeout,
+ Duration::from_secs(6)
+ );
+ assert_eq!(
+ effective_http_options_with_sync(&opts, false).timeout,
+ opts.http.timeout
+ );
+ opts.prom_store.with_metric_engine = false;
+ assert_eq!(
+ effective_http_options_with_sync(&opts, false).timeout,
+ opts.http.timeout
+ );
+ opts.prom_store.with_metric_engine = true;
+ opts.pending_rows_batcher
+ .logical_table
+ .as_mut()
+ .unwrap()
+ .protocols
+ .clear();
+ assert_eq!(
+ effective_http_options_with_sync(&opts, true).timeout,
+ opts.http.timeout
+ );
+ }
+
#[test]
fn test_effective_prom_batching_controls() {
// Only an enabled shared Prom selection replaces the legacy controls.
@@ -532,7 +607,7 @@ mod tests {
opts.prom_store.enable = prom_enabled;
opts.prom_store
.experimental_enable_prometheus_native_histogram = true;
- let shared = &mut opts.pending_rows_batcher;
+ let shared = &mut opts.pending_rows_batcher.table;
shared.protocols = vec![if selected {
BatchingProtocol::Prom
} else {
@@ -569,7 +644,7 @@ mod tests {
#[test]
fn test_http_timeout_covers_synchronous_batchers() {
- // Shared ordinary writes remain synchronous even when Prom is asynchronous.
+ // Only synchronous batchers extend the HTTP timeout.
for (
protocols,
metric_engine,
@@ -579,10 +654,10 @@ mod tests {
timeout_secs,
expected_secs,
) in [
- (vec![BatchingProtocol::Prom], false, false, 5, 2, 1, 6),
+ (vec![BatchingProtocol::Prom], false, false, 5, 2, 1, 1),
(vec![BatchingProtocol::Prom], true, false, 5, 2, 1, 1),
(vec![BatchingProtocol::Prom], true, true, 5, 2, 1, 6),
- (vec![BatchingProtocol::Influxdb], true, false, 5, 2, 1, 6),
+ (vec![BatchingProtocol::Influxdb], true, false, 5, 2, 1, 1),
(vec![BatchingProtocol::Influxdb], true, true, 5, 8, 1, 9),
(vec![BatchingProtocol::Influxdb], true, true, 8, 5, 1, 9),
(vec![BatchingProtocol::Influxdb], true, false, 5, 2, 0, 0),
@@ -595,8 +670,8 @@ mod tests {
opts.http.timeout = Duration::from_secs(timeout_secs);
opts.prom_store.with_metric_engine = metric_engine;
opts.prom_store.pending_rows_flush_interval = Duration::from_secs(legacy_secs);
- opts.pending_rows_batcher.protocols = protocols;
- opts.pending_rows_batcher.pending_rows_flush_interval =
+ opts.pending_rows_batcher.table.protocols = protocols;
+ opts.pending_rows_batcher.table.pending_rows_flush_interval =
Duration::from_secs(shared_secs);
assert_eq!(
effective_http_options_with_sync(&opts, batch_sync).timeout,
@@ -697,24 +772,30 @@ mod tests {
fn test_invalid_shared_batching_preserves_prom_store_options() {
type KnobMutator = fn(&mut FrontendOptions);
let cases: [KnobMutator; 5] = [
- |opts| opts.pending_rows_batcher.max_concurrent_flushes = usize::MAX,
- |opts| opts.pending_rows_batcher.worker_channel_capacity = usize::MAX,
- |opts| opts.pending_rows_batcher.max_inflight_requests = usize::MAX,
+ |opts| opts.pending_rows_batcher.table.max_concurrent_flushes = usize::MAX,
+ |opts| opts.pending_rows_batcher.table.worker_channel_capacity = usize::MAX,
+ |opts| opts.pending_rows_batcher.table.max_inflight_requests = usize::MAX,
|opts| {
- opts.pending_rows_batcher.flow_notification_queue_capacity =
- NonZeroUsize::new(usize::MAX).unwrap()
+ opts.pending_rows_batcher
+ .table
+ .flow_notification_queue_capacity = NonZeroUsize::new(usize::MAX).unwrap()
},
- |opts| opts.pending_rows_batcher.pending_rows_flush_interval = Duration::MAX,
+ |opts| opts.pending_rows_batcher.table.pending_rows_flush_interval = Duration::MAX,
];
for invalidate in cases {
let mut opts = FrontendOptions::default();
opts.http.timeout = Duration::from_secs(1);
opts.prom_store.pending_rows_flush_interval = Duration::from_secs(5);
- opts.pending_rows_batcher.protocols =
+ opts.pending_rows_batcher.table.protocols =
vec![BatchingProtocol::Prom, BatchingProtocol::Influxdb];
- opts.pending_rows_batcher.pending_rows_flush_interval = Duration::from_secs(10);
+ opts.pending_rows_batcher.table.pending_rows_flush_interval = Duration::from_secs(10);
invalidate(&mut opts);
- assert!(!opts.pending_rows_batcher.pending_rows_batching_enabled());
+ assert!(
+ !opts
+ .pending_rows_batcher
+ .table
+ .pending_rows_batching_enabled()
+ );
assert_eq!(opts.prom_store, effective_prom_store_options(&opts));
assert_eq!(
Duration::from_secs(6),
diff --git a/src/frontend/src/service_config.rs b/src/frontend/src/service_config.rs
index bcd1097cfcf..2c761c4242a 100644
--- a/src/frontend/src/service_config.rs
+++ b/src/frontend/src/service_config.rs
@@ -26,6 +26,6 @@ pub use jaeger::JaegerOptions;
pub use mysql::MysqlOptions;
pub use opentsdb::OpentsdbOptions;
pub use otlp::OtlpOptions;
-pub use pending_rows_batcher::PendingRowsBatcherOptions;
+pub use pending_rows_batcher::{BatcherOptions, PendingRowsBatcherOptions};
pub use postgres::PostgresOptions;
pub use prom_store::PromStoreOptions;
diff --git a/src/frontend/src/service_config/pending_rows_batcher.rs b/src/frontend/src/service_config/pending_rows_batcher.rs
index ef7ce14ef02..61572015d7c 100644
--- a/src/frontend/src/service_config/pending_rows_batcher.rs
+++ b/src/frontend/src/service_config/pending_rows_batcher.rs
@@ -20,10 +20,28 @@ use serde::{Deserialize, Serialize};
use servers::http::BatchingProtocol;
use tokio::sync::Semaphore;
-/// Experimental table write batching options shared by HTTP ingestion protocols.
-#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
+use crate::frontend::FrontendOptions;
+
+/// Independent ordinary-table and logical-table batching configuration.
+#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
pub struct PendingRowsBatcherOptions {
+ /// Existing ordinary-table controls retain their original TOML paths.
+ #[serde(flatten)]
+ pub table: BatcherOptions,
+ /// Independent logical-table controls; absence retains the legacy Prom fallback.
+ #[serde(
+ default,
+ skip_serializing_if = "Option::is_none",
+ deserialize_with = "deserialize_logical_options"
+ )]
+ pub logical_table: Option,
+}
+
+/// Write batching controls shared by HTTP ingestion protocols.
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(default)]
+pub struct BatcherOptions {
/// HTTP write protocols sharing this batcher; empty disables all entrances.
pub protocols: Vec,
/// Time from the first pending submission to a timed flush. Zero disables batching.
@@ -41,7 +59,7 @@ pub struct PendingRowsBatcherOptions {
pub flow_notification_queue_capacity: NonZeroUsize,
}
-impl PendingRowsBatcherOptions {
+impl BatcherOptions {
/// Returns whether a protocol opts in and its controls pass construction validation.
pub fn pending_rows_batching_enabled(&self) -> bool {
!self.protocols.is_empty()
@@ -58,79 +76,229 @@ impl PendingRowsBatcherOptions {
}
}
-impl Default for PendingRowsBatcherOptions {
+impl Default for BatcherOptions {
fn default() -> Self {
Self {
protocols: Vec::new(),
pending_rows_flush_interval: Duration::ZERO,
max_batch_rows: 100_000,
max_concurrent_flushes: 256,
- worker_channel_capacity: 65526,
+ worker_channel_capacity: 65_536,
max_inflight_requests: 3000,
flow_notification_queue_capacity: NonZeroUsize::new(1024).unwrap_or(NonZeroUsize::MIN),
}
}
}
+/// Rejects protocols that cannot write metric-engine logical tables.
+fn deserialize_logical_options<'de, D>(deserializer: D) -> Result, D::Error>
+where
+ D: serde::Deserializer<'de>,
+{
+ let options = Option::::deserialize(deserializer)?;
+ if options.as_ref().is_some_and(|options| {
+ options
+ .protocols
+ .iter()
+ .any(|protocol| !matches!(protocol, BatchingProtocol::Prom | BatchingProtocol::Otlp))
+ }) {
+ return Err(serde::de::Error::custom(
+ "pending_rows_batcher.logical_table only supports prom and otlp",
+ ));
+ }
+ Ok(options)
+}
+
+impl FrontendOptions {
+ /// Explicit configuration wins even when it disables batching.
+ pub(crate) fn table_batcher_options(&self) -> &BatcherOptions {
+ &self.pending_rows_batcher.table
+ }
+
+ /// Resolve legacy Prom controls once without merging fields across config blocks.
+ pub(crate) fn logical_batcher_options(&self) -> BatcherOptions {
+ if let Some(options) = &self.pending_rows_batcher.logical_table {
+ return options.clone();
+ }
+ let shared = &self.pending_rows_batcher.table;
+ if shared.protocols.contains(&BatchingProtocol::Prom)
+ && shared.pending_rows_batching_enabled()
+ {
+ let mut options = shared.clone();
+ options.protocols = vec![BatchingProtocol::Prom];
+ return options;
+ }
+ let prom = &self.prom_store;
+ BatcherOptions {
+ protocols: vec![BatchingProtocol::Prom],
+ pending_rows_flush_interval: prom.pending_rows_flush_interval,
+ max_batch_rows: prom.max_batch_rows,
+ max_concurrent_flushes: prom.max_concurrent_flushes,
+ worker_channel_capacity: prom.worker_channel_capacity,
+ max_inflight_requests: prom.max_inflight_requests,
+ flow_notification_queue_capacity: prom.flow_notification_queue_capacity,
+ }
+ }
+}
+
#[cfg(test)]
mod tests {
use crate::service_config::pending_rows_batcher::*;
+ #[test]
+ fn test_batcher_config_precedence() {
+ use crate::frontend::FrontendOptions;
+ let legacy = "[prom_store]\nenable = true\nwith_metric_engine = true\npending_rows_flush_interval = '2s'\nmax_batch_rows = 9\n";
+ let shared = "[pending_rows_batcher]\nprotocols = ['prom', 'otlp']\npending_rows_flush_interval = '3s'\n";
+ let options: FrontendOptions = toml::from_str(legacy).unwrap();
+ assert_eq!(options.logical_batcher_options().max_batch_rows, 9);
+ assert_eq!(
+ options.logical_batcher_options().protocols,
+ vec![BatchingProtocol::Prom]
+ );
+ let options: FrontendOptions = toml::from_str(&format!("{legacy}{shared}")).unwrap();
+ assert_eq!(
+ options
+ .logical_batcher_options()
+ .pending_rows_flush_interval,
+ Duration::from_secs(3)
+ );
+ assert_eq!(options.logical_batcher_options().max_batch_rows, 100_000);
+ assert_eq!(
+ options.logical_batcher_options().protocols,
+ vec![BatchingProtocol::Prom]
+ );
+ for explicit in [
+ "",
+ "protocols = ['otlp']\npending_rows_flush_interval = '5s'",
+ ] {
+ let options: FrontendOptions = toml::from_str(&format!(
+ "{legacy}{shared}[pending_rows_batcher.logical_table]\n{explicit}"
+ ))
+ .unwrap();
+ assert_eq!(
+ &options.logical_batcher_options(),
+ options.pending_rows_batcher.logical_table.as_ref().unwrap()
+ );
+ assert_eq!(
+ options.table_batcher_options(),
+ &toml::from_str::(&format!("{legacy}{shared}"))
+ .unwrap()
+ .pending_rows_batcher
+ .table
+ );
+ let restored: FrontendOptions =
+ toml::from_str(&toml::to_string(&options).unwrap()).unwrap();
+ assert_eq!(
+ options.pending_rows_batcher.logical_table,
+ restored.pending_rows_batcher.logical_table
+ );
+ assert_eq!(options.pending_rows_batcher, restored.pending_rows_batcher);
+ }
+ }
+
+ #[test]
+ fn test_batcher_config_independent_defaults() {
+ use crate::frontend::FrontendOptions;
+
+ let parent = "[pending_rows_batcher]\nprotocols = ['prom']\npending_rows_flush_interval = '3s'\nmax_batch_rows = 7\n";
+ let options: FrontendOptions = toml::from_str(parent).unwrap();
+ let serialized = toml::to_string(&options).unwrap();
+ assert!(!serialized.contains("[pending_rows_batcher.logical_table]"));
+ assert!(!serialized.contains("[pending_rows_batcher.table]"));
+ let restored: FrontendOptions = toml::from_str(&serialized).unwrap();
+ assert_eq!(restored.logical_batcher_options().max_batch_rows, 7);
+
+ let options: FrontendOptions =
+ toml::from_str(&format!("{parent}[pending_rows_batcher.logical_table]")).unwrap();
+ assert_eq!(options.table_batcher_options().max_batch_rows, 7);
+ assert_eq!(options.logical_batcher_options(), BatcherOptions::default());
+ assert!(
+ !options
+ .logical_batcher_options()
+ .pending_rows_batching_enabled()
+ );
+ }
+
+ #[test]
+ fn test_logical_batcher_config_protocols() {
+ use crate::frontend::FrontendOptions;
+ for protocols in ["[]", "['prom']", "['otlp']", "['prom', 'otlp']"] {
+ let options: FrontendOptions = toml::from_str(&format!(
+ "[pending_rows_batcher.logical_table]\nprotocols = {protocols}"
+ ))
+ .unwrap();
+ assert!(options.pending_rows_batcher.logical_table.is_some());
+ }
+ for protocol in [
+ "influxdb",
+ "logs",
+ "loki",
+ "http_sql",
+ "opentsdb",
+ "elasticsearch",
+ "splunk",
+ ] {
+ assert!(
+ toml::from_str::(&format!(
+ "[pending_rows_batcher.logical_table]\nprotocols = ['{protocol}']"
+ ))
+ .is_err()
+ );
+ }
+ }
+
#[test]
fn test_protocols() {
- let options: PendingRowsBatcherOptions = toml::from_str(
+ let options: BatcherOptions = toml::from_str(
"protocols = ['influxdb', 'opentsdb', 'otlp', 'logs', 'loki', 'splunk', 'elasticsearch', 'http_sql', 'prom']",
).unwrap();
assert_eq!(options.protocols.len(), 9);
assert!(options.protocols.contains(&BatchingProtocol::HttpSql));
- assert!(PendingRowsBatcherOptions::default().protocols.is_empty());
+ assert!(BatcherOptions::default().protocols.is_empty());
for invalid in ["sql", "jaeger", "unknown"] {
assert!(
- toml::from_str::(&format!("protocols = ['{invalid}']"))
- .is_err()
+ toml::from_str::(&format!("protocols = ['{invalid}']")).is_err()
);
}
}
#[test]
fn test_notification_capacity() {
- let default = PendingRowsBatcherOptions::default();
+ let default = BatcherOptions::default();
assert_eq!(default.flow_notification_queue_capacity.get(), 1024);
- let configured: PendingRowsBatcherOptions =
+ let configured: BatcherOptions =
toml::from_str("flow_notification_queue_capacity = 8").unwrap();
assert_eq!(configured.flow_notification_queue_capacity.get(), 8);
- assert!(
- toml::from_str::("flow_notification_queue_capacity = 0")
- .is_err()
- );
+ assert!(toml::from_str::("flow_notification_queue_capacity = 0").is_err());
}
#[test]
fn test_defaults_and_roundtrip() {
- let options: PendingRowsBatcherOptions = toml::from_str("").unwrap();
- assert_eq!(options, PendingRowsBatcherOptions::default());
+ let options: BatcherOptions = toml::from_str("").unwrap();
+ assert_eq!(options, BatcherOptions::default());
assert!(!options.pending_rows_batching_enabled());
assert_eq!(options.max_batch_rows, 100_000);
assert_eq!(options.max_concurrent_flushes, 256);
- assert_eq!(options.worker_channel_capacity, 65526);
+ assert_eq!(options.worker_channel_capacity, 65_536);
assert_eq!(options.max_inflight_requests, 3000);
let serialized = toml::to_string(&options).unwrap();
assert_eq!(
options,
- toml::from_str::(&serialized).unwrap()
+ toml::from_str::(&serialized).unwrap()
);
}
#[test]
fn test_partial_options_and_zero_controls() {
- let options: PendingRowsBatcherOptions =
+ let options: BatcherOptions =
toml::from_str("pending_rows_flush_interval = '5ms'").unwrap();
assert_eq!(
options.pending_rows_flush_interval,
Duration::from_millis(5)
);
assert!(!options.pending_rows_batching_enabled());
- let enabled: PendingRowsBatcherOptions =
+ let enabled: BatcherOptions =
toml::from_str("protocols = ['http_sql']\npending_rows_flush_interval = '5ms'")
.unwrap();
assert!(enabled.pending_rows_batching_enabled());
@@ -140,7 +308,7 @@ mod tests {
"worker_channel_capacity",
"max_inflight_requests",
] {
- let options: PendingRowsBatcherOptions = toml::from_str(&format!(
+ let options: BatcherOptions = toml::from_str(&format!(
"protocols = ['influxdb']\npending_rows_flush_interval = '5ms'\n{field} = 0"
))
.unwrap();
diff --git a/src/frontend/src/service_config/prom_store.rs b/src/frontend/src/service_config/prom_store.rs
index f8429a86925..2fdd05043db 100644
--- a/src/frontend/src/service_config/prom_store.rs
+++ b/src/frontend/src/service_config/prom_store.rs
@@ -54,7 +54,7 @@ fn default_max_concurrent_flushes() -> usize {
}
fn default_worker_channel_capacity() -> usize {
- 65526
+ 65_536
}
fn default_max_inflight_requests() -> usize {
diff --git a/src/operator/src/batcher.rs b/src/operator/src/batcher.rs
index 4c5e9892c0b..677524d53f3 100644
--- a/src/operator/src/batcher.rs
+++ b/src/operator/src/batcher.rs
@@ -29,7 +29,8 @@ pub trait PendingRowsBatcher: Send + Sync {
/// Acquires one slot per original request, shared by all of its table submissions.
async fn acquire(&self) -> Result>;
- /// Waits for the submitted rows to be written, retaining the slot through completion.
+ /// Submits rows according to the acknowledgement policy, retaining the slot until
+ /// writing completes even when the response acknowledges queue admission only.
/// Cancelling the response wait does not retract an already enqueued submission.
async fn submit(
&self,
diff --git a/src/operator/src/insert.rs b/src/operator/src/insert.rs
index afc903621e2..49215f89713 100644
--- a/src/operator/src/insert.rs
+++ b/src/operator/src/insert.rs
@@ -156,6 +156,127 @@ pub struct InstantAndNormalInsertRequests {
}
impl Inserter {
+ /// Checks the assumptions of the logical bulk path without changing tables.
+ /// Unsupported requests retain ordinary insertion, including schema policy,
+ /// defaults, instant TTL and row-based Flow delivery.
+ pub async fn can_batch_metric_rows(
+ &self,
+ requests: &RowInsertRequests,
+ ctx: &QueryContextRef,
+ physical_table: &str,
+ ) -> Result {
+ if self.auto_create_disabled_reason(ctx)?.is_some() || ctx.extension(TTL_KEY).is_some() {
+ return Ok(false);
+ }
+ for request in &requests.inserts {
+ // The logical bulk encoder only supports scalar metric schemas.
+ // Check new tables too, before catalog lookup or schema changes.
+ if request.rows.as_ref().is_some_and(|rows| {
+ rows.schema.iter().any(|column| {
+ column.datatype_extension.is_some()
+ || !matches!(
+ ColumnDataType::try_from(column.datatype),
+ Ok(ColumnDataType::TimestampMillisecond
+ | ColumnDataType::Float64
+ | ColumnDataType::String)
+ )
+ })
+ }) {
+ return Ok(false);
+ }
+
+ let Some(table) = self
+ .get_table(
+ ctx.current_catalog(),
+ &ctx.current_schema(),
+ &request.table_name,
+ )
+ .await?
+ else {
+ continue;
+ };
+ let info = table.table_info();
+ if info.meta.engine != METRIC_ENGINE_NAME
+ || info.is_ttl_instant_table()
+ || info
+ .meta
+ .options
+ .extra_options
+ .get(LOGICAL_TABLE_METADATA_KEY)
+ .map(String::as_str)
+ != Some(physical_table)
+ || info
+ .meta
+ .schema
+ .column_schemas()
+ .iter()
+ .any(|column| column.default_constraint().is_some())
+ {
+ return Ok(false);
+ }
+ // Physical metric tags are nullable even when their logical schema
+ // is not. Arrow alignment is stricter than ordinary metric insertion.
+ if info
+ .meta
+ .primary_key_indices
+ .iter()
+ .any(|&index| !info.meta.schema.column_schemas()[index].is_nullable())
+ {
+ return Ok(false);
+ }
+ // The current Flow cache does not distinguish streaming and batch
+ // flows. Keep all Flow sources on the row-based delivery path.
+ match self.table_flownode_set_cache.get(info.table_id()).await {
+ Ok(None) => {}
+ Ok(Some(flows)) if flows.is_empty() => {}
+ _ => return Ok(false),
+ }
+ }
+ Ok(true)
+ }
+
+ /// Meters an original logical-table request before bulk routing, without
+ /// cloning its rows or changing the request boundary used for accounting.
+ pub async fn meter_row_inserts(
+ requests: &mut RowInsertRequests,
+ ctx: &QueryContextRef,
+ ) -> Result {
+ let metered = InstantAndNormalInsertRequests {
+ normal_requests: RegionInsertRequests {
+ requests: requests
+ .inserts
+ .iter_mut()
+ .map(|request| RegionInsertRequest {
+ rows: request.rows.take(),
+ ..Default::default()
+ })
+ .collect(),
+ },
+ instant_requests: RegionInsertRequests::default(),
+ };
+ let cost = write_meter!(
+ ctx.current_catalog(),
+ ctx.current_schema(),
+ metered,
+ ctx.write_rows_to_admit(
+ ctx.current_catalog(),
+ &ctx.current_schema(),
+ count_insert_rows(&metered)?
+ ),
+ ctx.channel() as u8
+ )
+ .await
+ .context(WriteRejectedSnafu);
+ for (request, region) in requests
+ .inserts
+ .iter_mut()
+ .zip(metered.normal_requests.requests)
+ {
+ request.rows = region.rows;
+ }
+ cost
+ }
+
pub fn new(
catalog_manager: CatalogManagerRef,
partition_manager: PartitionRuleManagerRef,
@@ -2313,19 +2434,67 @@ mod tests {
);
assert_eq!(ctx.channel(), Channel::Prometheus);
}
- let attempts = meter.attempts.lock().unwrap();
- let totals = attempts
+ {
+ let attempts = meter.attempts.lock().unwrap();
+ let totals = attempts
+ .iter()
+ .map(|r| (r.schema.as_str(), r.rows, r.value))
+ .collect::>();
+ assert_eq!(
+ totals,
+ if enabled {
+ vec![("a", 4, 0), ("b", 2, 0)]
+ } else {
+ vec![]
+ }
+ );
+ }
+ meter.attempts.lock().unwrap().clear();
+ meter.reject.store(false, Ordering::Relaxed);
+ let ctx = Arc::new(QueryContext::with_channel(
+ CATALOG,
+ "logical",
+ Channel::Otlp,
+ ));
+ let admitted = admit_write(2, &ctx).await.unwrap();
+ let mut requests = RowInsertRequests {
+ inserts: vec![RowInsertRequest {
+ table_name: "metric".to_string(),
+ rows: Some(Rows {
+ schema: vec![],
+ rows: vec![api::v1::Row::default(); 2],
+ }),
+ }],
+ };
+ let original = requests.clone();
+ let cost = Inserter::meter_row_inserts(&mut requests, &admitted)
+ .await
+ .unwrap();
+ assert_eq!(cost, if enabled { 17 } else { 0 });
+ assert_eq!(requests, original);
+ let records = meter
+ .attempts
+ .lock()
+ .unwrap()
.iter()
- .map(|r| (r.schema.as_str(), r.rows, r.value))
+ .map(|record| (record.rows, record.value))
.collect::>();
assert_eq!(
- totals,
+ records,
if enabled {
- vec![("a", 4, 0), ("b", 2, 0)]
+ vec![(2, 0), (0, 17)]
} else {
vec![]
}
);
+ meter.reject.store(true, Ordering::Relaxed);
+ let result = Inserter::meter_row_inserts(&mut requests, &admitted).await;
+ if enabled {
+ assert_eq!(result.unwrap_err().status_code(), StatusCode::RateLimited);
+ } else {
+ assert_eq!(result.unwrap(), 0);
+ }
+ assert_eq!(requests, original);
}
#[test]
@@ -2570,6 +2739,142 @@ mod tests {
)
}
+ #[tokio::test]
+ async fn test_logical_batcher_eligibility() {
+ use catalog::RegisterTableRequest;
+ use catalog::memory::MemoryCatalogManager;
+ use common_meta::instruction::{CacheIdent, CreateFlow};
+ use common_meta::kv_backend::KvBackendRef;
+ use common_meta::kv_backend::memory::MemoryKvBackend;
+ use datatypes::schema::{ColumnDefaultConstraint, SchemaBuilder};
+ let requests = RowInsertRequests {
+ inserts: vec![RowInsertRequest {
+ table_name: "test_table".to_string(),
+ rows: None,
+ }],
+ };
+ let original =
+ make_table_ref_with_schema("ts", "value", ConcreteDataType::float64_datatype())
+ .table_info();
+ for case in [
+ "eligible",
+ "physical",
+ "ordinary",
+ "instant",
+ "disabled",
+ "hint",
+ "flow",
+ "required_tag",
+ "default",
+ ] {
+ let mut info = (*original).clone();
+ info.meta.engine = METRIC_ENGINE_NAME.to_string();
+ info.meta.options.extra_options.insert(
+ LOGICAL_TABLE_METADATA_KEY.to_string(),
+ "physical".to_string(),
+ );
+ let mut ctx = QueryContext::arc().fork();
+ match case {
+ "physical" => {
+ info.meta
+ .options
+ .extra_options
+ .insert(LOGICAL_TABLE_METADATA_KEY.to_string(), "other".to_string());
+ }
+ "ordinary" => info.meta.engine = "mito".to_string(),
+ "instant" => info.meta.options.ttl = Some(common_time::ttl::TimeToLive::Instant),
+ "hint" => ctx.set_extension(AUTO_CREATE_TABLE_KEY, "false"),
+ "required_tag" | "default" => {
+ let mut columns = info.meta.schema.column_schemas().to_vec();
+ let mut tag = ColumnSchema::new(
+ "tag",
+ ConcreteDataType::string_datatype(),
+ case != "required_tag",
+ );
+ if case == "default" {
+ tag = tag
+ .with_default_constraint(Some(ColumnDefaultConstraint::null_value()))
+ .unwrap();
+ }
+ columns.push(tag);
+ info.meta.schema = Arc::new(
+ SchemaBuilder::try_from_columns(columns)
+ .unwrap()
+ .build()
+ .unwrap(),
+ );
+ info.meta.primary_key_indices = vec![2];
+ }
+ _ => {}
+ }
+ let catalog = MemoryCatalogManager::with_default_setup();
+ let table = Arc::new(table::Table::new(
+ Arc::new(info),
+ table::metadata::FilterPushDownType::Unsupported,
+ Arc::new(DummyDataSource),
+ ));
+ catalog
+ .register_table_sync(RegisterTableRequest {
+ catalog: DEFAULT_CATALOG_NAME.to_string(),
+ schema: DEFAULT_SCHEMA_NAME.to_string(),
+ table_name: "test_table".to_string(),
+ table_id: 1,
+ table,
+ })
+ .unwrap();
+ let mut inserter = batcher_test_inserter().await;
+ inserter.catalog_manager = catalog;
+ inserter.auto_create_table = case != "disabled";
+ let kv_backend: KvBackendRef = Arc::new(MemoryKvBackend::default());
+ inserter.table_flownode_set_cache = Arc::new(new_table_flownode_set_cache(
+ String::new(),
+ Cache::new(10),
+ kv_backend,
+ ));
+ if case == "flow" {
+ inserter
+ .table_flownode_set_cache
+ .invalidate(&[CacheIdent::CreateFlow(CreateFlow {
+ flow_id: 1,
+ source_table_ids: vec![1],
+ partition_to_peer_mapping: vec![(0, Peer::empty(1))],
+ })])
+ .await
+ .unwrap();
+ }
+ assert_eq!(
+ inserter
+ .can_batch_metric_rows(&requests, &Arc::new(ctx), "physical")
+ .await
+ .unwrap(),
+ case == "eligible",
+ "{case}"
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn test_batcher_meter_preserves_request() {
+ let mut requests = RowInsertRequests {
+ inserts: vec![RowInsertRequest {
+ table_name: "sample".to_string(),
+ rows: Some(Rows {
+ schema: vec![],
+ rows: vec![api::v1::Row {
+ values: vec![Value {
+ value_data: Some(api::v1::value::ValueData::F64Value(1.5)),
+ }],
+ }],
+ }),
+ }],
+ };
+ let expected = requests.clone();
+ Inserter::meter_row_inserts(&mut requests, &QueryContext::arc())
+ .await
+ .unwrap();
+ assert_eq!(requests, expected);
+ }
+
#[tokio::test]
async fn test_instant_table_bypasses_batcher() {
let batcher: Arc = Arc::new(UnexpectedBatcher);
diff --git a/src/servers/src/batcher.rs b/src/servers/src/batcher.rs
index 10f8f558eb3..6b6ca8f7ee7 100644
--- a/src/servers/src/batcher.rs
+++ b/src/servers/src/batcher.rs
@@ -21,3 +21,21 @@ pub mod table;
#[cfg(test)]
mod test_util;
+
+/// Controls whether batching waits for storage before replying to the client.
+const PENDING_ROWS_BATCH_SYNC_ENV: &str = "PENDING_ROWS_BATCH_SYNC";
+
+/// Returns whether pending-row batch submissions wait for the flush result
+/// before replying to the client (synchronous mode), controlled by the
+/// `PENDING_ROWS_BATCH_SYNC` environment variable and defaulting to `true`.
+///
+/// Callers that reason about how long a remote write request may block (e.g.
+/// the frontend HTTP timeout fallback) must consult this instead of
+/// duplicating the env lookup.
+pub fn pending_rows_batch_sync_enabled() -> bool {
+ std::env::var(PENDING_ROWS_BATCH_SYNC_ENV)
+ .ok()
+ .as_deref()
+ .and_then(|v| v.parse::().ok())
+ .unwrap_or(true)
+}
diff --git a/src/servers/src/batcher/logical_table.rs b/src/servers/src/batcher/logical_table.rs
index f4c764ee8a8..bd7a26da613 100644
--- a/src/servers/src/batcher/logical_table.rs
+++ b/src/servers/src/batcher/logical_table.rs
@@ -21,6 +21,7 @@ mod tables;
#[cfg(test)]
mod test_util;
+use std::future::{Future, ready};
use std::num::NonZeroUsize;
use std::sync::Arc;
use std::time::Duration;
@@ -54,6 +55,7 @@ pub use crate::batcher::logical_table::region_write::{
pub use crate::batcher::logical_table::tables::{
PendingRowsSchemaAlterer, PendingRowsSchemaAltererRef,
};
+use crate::batcher::pending_rows_batch_sync_enabled;
use crate::error;
use crate::error::{Error, Result};
use crate::metrics::{
@@ -62,24 +64,6 @@ use crate::metrics::{
const PHYSICAL_TABLE_KEY: &str = "physical_table";
-/// Whether wait for ingestion result before reply to client.
-const PENDING_ROWS_BATCH_SYNC_ENV: &str = "PENDING_ROWS_BATCH_SYNC";
-
-/// Returns whether pending-row batch submissions wait for the flush result
-/// before replying to the client (synchronous mode), controlled by the
-/// `PENDING_ROWS_BATCH_SYNC` environment variable and defaulting to `true`.
-///
-/// Callers that reason about how long a remote write request may block (e.g.
-/// the frontend HTTP timeout fallback) must consult this instead of
-/// duplicating the env lookup.
-pub fn pending_rows_batch_sync_enabled() -> bool {
- std::env::var(PENDING_ROWS_BATCH_SYNC_ENV)
- .ok()
- .as_deref()
- .and_then(|v| v.parse::().ok())
- .unwrap_or(true)
-}
-
const WORKER_IDLE_TIMEOUT_MULTIPLIER: u32 = 3;
#[derive(Debug, Clone, Hash, Eq, PartialEq)]
@@ -182,14 +166,31 @@ impl LogicalTablePendingRowsBatcher {
impl LogicalTablePendingRowsBatcher {
pub async fn submit(&self, requests: RowInsertRequests, ctx: QueryContextRef) -> Result {
+ self.submit_with(requests, ctx, |_| ready(Ok(())))
+ .await
+ .map(|(rows, ())| rows)
+ }
+
+ /// Submits with request-level accounting after schema preparation and before
+ /// queue admission. Acknowledgement follows the global batching policy.
+ pub async fn submit_with(
+ &self,
+ requests: RowInsertRequests,
+ ctx: QueryContextRef,
+ after_prepare: impl FnOnce(RowInsertRequests) -> F,
+ ) -> Result<(u64, T)>
+ where
+ F: Future