From 94c19d87804b8f144a433a292e67e567ded35caf Mon Sep 17 00:00:00 2001 From: WenyXu Date: Tue, 22 Sep 2026 06:43:13 +0000 Subject: [PATCH] fix: unify pending rows batch acknowledgement policy Signed-off-by: WenyXu --- config/config.md | 4 +- config/frontend.example.toml | 2 + config/standalone.example.toml | 2 + src/frontend/src/instance/otlp.rs | 8 +- src/frontend/src/server.rs | 26 ++--- src/operator/src/batcher.rs | 3 +- src/servers/src/batcher.rs | 18 ++++ src/servers/src/batcher/logical_table.rs | 48 ++-------- src/servers/src/batcher/table.rs | 117 ++++++++++++++++++++++- src/servers/src/batcher/table/batch.rs | 2 + tests-integration/src/otlp.rs | 34 +++++-- tests-integration/src/standalone.rs | 1 + 12 files changed, 196 insertions(+), 69 deletions(-) diff --git a/config/config.md b/config/config.md index 29deeb15234..694359713eb 100644 --- a/config/config.md +++ b/config/config.md @@ -75,7 +75,7 @@ | `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` | -- | -- | Ordinary-table batching for opted-in HTTP ingestion protocols.
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` | -- | -- | 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. | @@ -343,7 +343,7 @@ | `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` | -- | -- | Ordinary-table batching for opted-in HTTP ingestion protocols.
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` | -- | -- | 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. | diff --git a/config/frontend.example.toml b/config/frontend.example.toml index 2e9ee1d4e93..4263c0bc7c3 100644 --- a/config/frontend.example.toml +++ b/config/frontend.example.toml @@ -232,6 +232,8 @@ enable = true default_merge_mode = "last_non_null" ## 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] diff --git a/config/standalone.example.toml b/config/standalone.example.toml index 813a121a630..aa6f894d38e 100644 --- a/config/standalone.example.toml +++ b/config/standalone.example.toml @@ -211,6 +211,8 @@ enable = true default_merge_mode = "last_non_null" ## 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] diff --git a/src/frontend/src/instance/otlp.rs b/src/frontend/src/instance/otlp.rs index 231fc874880..7a0f706c58d 100644 --- a/src/frontend/src/instance/otlp.rs +++ b/src/frontend/src/instance/otlp.rs @@ -189,7 +189,7 @@ impl OpenTelemetryProtocolHandler for Instance { // OTLP tables have one sample field in both the legacy and physical paths. let output = if let Some(batcher) = batcher { let (rows, cost) = batcher - .submit_sync(requests, ctx.clone(), |mut requests| { + .submit_with(requests, ctx.clone(), |mut requests| { let ctx = ctx.clone(); async move { Inserter::meter_row_inserts(&mut requests, &ctx) @@ -216,9 +216,9 @@ impl OpenTelemetryProtocolHandler for Instance { }; 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 c07b57bafb2..0a6612ba31c 100644 --- a/src/frontend/src/server.rs +++ b/src/frontend/src/server.rs @@ -24,7 +24,7 @@ use common_base::Plugins; use common_config::Configurable; use common_telemetry::{info, warn}; use meta_client::MetaClientOptions; -use servers::batcher::logical_table::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; @@ -449,9 +449,8 @@ fn effective_http_options_with_sync(opts: &FrontendOptions, batch_sync: bool) -> let mut http = opts.http.clone(); let prom_store = effective_prom_store_options(opts); let shared = opts.table_batcher_options(); - // 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 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) @@ -460,7 +459,8 @@ fn effective_http_options_with_sync(opts: &FrontendOptions, batch_sync: bool) -> let prom_interval = (prom_store.pending_rows_batching_enabled() && batch_sync) .then_some(prom_store.pending_rows_flush_interval); let logical = opts.logical_batcher_options(); - let otlp_interval = (opts.otlp.enable + let otlp_interval = (batch_sync + && opts.otlp.enable && opts.prom_store.with_metric_engine && logical.protocols.contains(&BatchingProtocol::Otlp) && logical.pending_rows_batching_enabled()) @@ -561,13 +561,17 @@ mod tests { ..Default::default() }); assert!(!effective_prom_store_options(&opts).pending_rows_batching_enabled()); - // OTLP waits for storage even when Prom acknowledges asynchronously, - // and does not depend on enabling the Prom HTTP endpoint. + // 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, false).timeout, + 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, @@ -640,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, @@ -650,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), 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/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 a340e76e3df..bd7a26da613 100644 --- a/src/servers/src/batcher/logical_table.rs +++ b/src/servers/src/batcher/logical_table.rs @@ -55,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::{ @@ -63,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)] @@ -183,34 +166,19 @@ impl LogicalTablePendingRowsBatcher { impl LogicalTablePendingRowsBatcher { pub async fn submit(&self, requests: RowInsertRequests, ctx: QueryContextRef) -> Result { - self.submit_inner(requests, ctx, self.pending_rows_batch_sync, |_| { - ready(Ok(())) - }) - .await - .map(|(rows, ())| rows) + self.submit_with(requests, ctx, |_| ready(Ok(()))) + .await + .map(|(rows, ())| rows) } - /// Waits for storage regardless of the Prom acknowledgement mode. The hook - /// runs after schema preparation, before admission, for request-level accounting. - pub async fn submit_sync( + /// 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> + Send, - { - self.submit_inner(requests, ctx, true, after_prepare).await - } - - async fn submit_inner( - &self, - requests: RowInsertRequests, - ctx: QueryContextRef, - wait: bool, - after_prepare: impl FnOnce(RowInsertRequests) -> F, - ) -> Result<(u64, T)> where F: Future> + Send, { @@ -291,7 +259,7 @@ impl LogicalTablePendingRowsBatcher { } } - if wait { + if self.pending_rows_batch_sync { let result = { let _timer = PENDING_ROWS_BATCH_INGEST_STAGE_ELAPSED .with_label_values(&["submit_wait_flush_result"]) diff --git a/src/servers/src/batcher/table.rs b/src/servers/src/batcher/table.rs index 1a46a7f10f3..67b3431a810 100644 --- a/src/servers/src/batcher/table.rs +++ b/src/servers/src/batcher/table.rs @@ -38,6 +38,7 @@ use snafu::ResultExt; use table::metadata::TableInfoRef; use tokio::sync::{OwnedSemaphorePermit, Semaphore, broadcast, oneshot}; +use crate::batcher::pending_rows_batch_sync_enabled; use crate::batcher::table::flow_notifier::FlowNotifier; use crate::batcher::table::metrics::PENDING_WORKERS; use crate::batcher::table::pending_batch::PendingBatch; @@ -70,6 +71,7 @@ pub struct TablePendingRowsBatcher { flush_limiter: FlushLimiter, request_limiter: RequestLimiter, worker_channel_capacity: usize, + pending_rows_batch_sync: bool, worker_idle_timeout: Duration, inserter: Arc, flow_notifier: FlowNotifier, @@ -110,6 +112,7 @@ impl TablePendingRowsBatcher { flush_limiter, request_limiter, worker_channel_capacity, + pending_rows_batch_sync: pending_rows_batch_sync_enabled(), worker_idle_timeout: flush_interval.checked_mul(3).unwrap_or(flush_interval), inserter, flow_notifier, @@ -160,8 +163,8 @@ impl PendingRowsBatcher for TablePendingRowsBatcher { }) } - /// Waits for completed bulk writes. Cancellation does not retract an admitted - /// submission. Combined failures affect all waiters and may be partial writes. + /// Acknowledges according to the global batching policy. Cancellation does + /// not retract an admitted submission. Flush failures may be partial writes. async fn submit( &self, table_info: TableInfoRef, @@ -210,6 +213,9 @@ impl PendingRowsBatcher for TablePendingRowsBatcher { } .fail(); } + if !self.pending_rows_batch_sync { + return Ok(total_rows); + } response_rx .await .map_err(|_| { @@ -236,7 +242,7 @@ mod tests { use api::v1::region::{RegionRequest, bulk_insert_request, region_request}; use api::v1::value::ValueData; use api::v1::{ColumnDataType, Row, Rows, Value}; - use arrow::array::{Int32Array, TimestampMillisecondArray}; + use arrow::array::{ArrayRef, Int32Array, TimestampMillisecondArray}; use arrow::datatypes::Schema as ArrowSchema; use arrow::record_batch::RecordBatch; use catalog::memory::MemoryCatalogManager; @@ -320,6 +326,111 @@ mod tests { } } + #[tokio::test] + async fn test_acknowledgement_preserves_admission() { + use operator::error::UnexpectedSnafu; + + use crate::batcher::pending_rows_batch_sync_enabled; + use crate::batcher::table::batch_key_from_ctx; + use crate::batcher::table::pending_batch::notify_batches; + + for sync in [false, true] { + for fail in [false, true] { + let backend = prepare_mocked_backend().await; + let nodes = Arc::new(MockDatanodeManager::new(BulkHandler { + requests: Arc::new(Mutex::new(Vec::new())), + report_missing_row: false, + expected_skip_wal: false, + expected_schema: "public".to_string(), + })); + let inserter = Arc::new(Inserter::new( + MemoryCatalogManager::new(), + create_partition_rule_manager(backend).await, + nodes, + mock_table_flownode_cache(1, vec![]).await, + true, + )); + let mut batcher = TablePendingRowsBatcher::try_new( + Duration::from_secs(3600), + 1, + 1, + 1, + 1, + NonZeroUsize::new(1).unwrap(), + inserter, + ) + .unwrap(); + assert_eq!( + batcher.pending_rows_batch_sync, + pending_rows_batch_sync_enabled() + ); + Arc::get_mut(&mut batcher).unwrap().pending_rows_batch_sync = sync; + let table = Arc::new(new_test_table_info(1, "ack", [0].into_iter())); + let ctx = QueryContext::arc(); + let key = batch_key_from_ctx(&table.name, &ctx); + // Hold the worker command to control completion independently of scheduling. + let (_, receiver) = batcher.workers.get_or_create(key, 1).await; + let mut receiver = receiver.unwrap(); + let batch = RecordBatch::try_from_iter(vec![( + "a", + Arc::new(Int32Array::from(vec![1])) as ArrayRef, + )]) + .unwrap(); + let permit = batcher.acquire().await.unwrap(); + let submitter = batcher.clone(); + let submitted = + tokio::spawn(async move { submitter.submit(table, batch, ctx, permit).await }); + let WorkerCommand::Submit(pending) = + timeout(Duration::from_secs(5), receiver.recv()) + .await + .unwrap() + .unwrap(); + let mut submitted = Some(submitted); + if sync { + assert!(!submitted.as_ref().unwrap().is_finished()); + } else { + assert_eq!( + timeout(Duration::from_secs(5), submitted.take().unwrap()) + .await + .unwrap() + .unwrap() + .unwrap(), + 1 + ); + } + // Early acknowledgement must not release capacity before completion. + let acquire = batcher.acquire(); + tokio::pin!(acquire); + assert!(futures::poll!(acquire.as_mut()).is_pending()); + let result = if fail { + Err(Arc::new( + UnexpectedSnafu { + violated: "flush failed".to_string(), + } + .build(), + )) + } else { + Ok(()) + }; + notify_batches(vec![pending], result); + if let Some(submitted) = submitted { + let result = timeout(Duration::from_secs(5), submitted) + .await + .unwrap() + .unwrap(); + assert_eq!(result.is_err(), fail); + if !fail { + assert_eq!(result.unwrap(), 1); + } + } + timeout(Duration::from_secs(5), acquire) + .await + .unwrap() + .unwrap(); + } + } + } + fn rows(value: i32) -> Rows { Rows { schema: vec![ diff --git a/src/servers/src/batcher/table/batch.rs b/src/servers/src/batcher/table/batch.rs index 21ebb94a33a..e9ab339a6ac 100644 --- a/src/servers/src/batcher/table/batch.rs +++ b/src/servers/src/batcher/table/batch.rs @@ -17,6 +17,7 @@ use std::time::Instant; use arrow::compute::concat_batches; use arrow::record_batch::RecordBatch; +use common_telemetry::error; use operator::error::{ComputeArrowSnafu, Result, UnexpectedSnafu}; use operator::insert::Inserter; use operator::metrics::DIST_INGEST_ROW_COUNT; @@ -55,6 +56,7 @@ pub(in crate::batcher::table) async fn flush_batch( notifier.notify(table, &combined); } Err(error) => { + error!(error; "Failed to flush table batch, rows: {}", batch.total_rows); FLUSH_FAILURES.inc(); FLUSH_DROPPED_ROWS.inc_by(batch.total_rows as u64); notify_batches(batch.submissions, Err(Arc::new(error))); diff --git a/tests-integration/src/otlp.rs b/tests-integration/src/otlp.rs index 6482b917bc2..338b03ff68f 100644 --- a/tests-integration/src/otlp.rs +++ b/tests-integration/src/otlp.rs @@ -514,12 +514,13 @@ WITH( ExponentialHistogram, ExponentialHistogramDataPoint, exponential_histogram_data_point, }; use prost::Message; + use servers::batcher::pending_rows_batch_sync_enabled; use servers::http::BatchingProtocol; use servers::http::test_helpers::TestClient; use session::protocol_ctx::{OtlpMetricCtx, ProtocolCtx}; // Fixed workload, only the batcher opt-in changes. Compare stored rows, - // schema evolution and immediate post-response visibility. + // schema evolution and visibility under the configured acknowledgement policy. let mut results = Vec::new(); for enabled in [false, true] { let standalone = GreptimeDbStandaloneBuilder::new(&format!("otlp_logical_{enabled}")) @@ -578,16 +579,33 @@ WITH( if enabled { assert_eq!( submissions.get_sample_count() - before, - 2, - "must exercise synchronous bulk submission" + if pending_rows_batch_sync_enabled() { + 2 + } else { + 0 + }, + "logical submissions must follow the global acknowledgement policy" ); } let sql = "SELECT greptime_timestamp, greptime_value, stream, extra FROM batch_alignment_total ORDER BY greptime_timestamp"; - let output = instance.do_query(sql, ctx.clone()).await.remove(0).unwrap(); - let OutputData::Stream(stream) = output.data else { - panic!("expected stream") - }; - let batches = RecordBatches::try_collect(stream).await.unwrap(); + let batches = tokio::time::timeout(Duration::from_secs(5), async { + loop { + let output = instance.do_query(sql, ctx.clone()).await.remove(0).unwrap(); + let OutputData::Stream(stream) = output.data else { + panic!("expected stream") + }; + let batches = RecordBatches::try_collect(stream).await.unwrap(); + if !enabled + || pending_rows_batch_sync_enabled() + || batches.iter().map(|batch| batch.num_rows()).sum::() == 2 + { + break batches; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); assert_eq!( batches.iter().map(|batch| batch.num_rows()).sum::(), 2 diff --git a/tests-integration/src/standalone.rs b/tests-integration/src/standalone.rs index 642741ee7f9..38f546e9ccb 100644 --- a/tests-integration/src/standalone.rs +++ b/tests-integration/src/standalone.rs @@ -121,6 +121,7 @@ impl GreptimeDbStandaloneBuilder { self } + /// Configures logical-table batching for integration tests. #[must_use] pub fn with_logical_batcher(mut self, options: BatcherOptions) -> Self { self.logical_batcher = Some(options);