diff --git a/.github/scripts/query-regression-run.py b/.github/scripts/query-regression-run.py index 62d11f4420..1aa8482537 100644 --- a/.github/scripts/query-regression-run.py +++ b/.github/scripts/query-regression-run.py @@ -31,6 +31,7 @@ DEFAULT_CASES = [ "tests/perf/query_cases/prom_remote_write_mixed_every/case.toml", "tests/perf/query_cases/prom_remote_write_integer_counter/case.toml", "tests/perf/query_cases/promql_pushdown_7913/case.toml", + "tests/perf/query_cases/analyze_verbose_many_files/case.toml", "tests/perf/query_cases/sql_topk_order_by/case.toml", "tests/perf/query_cases/sql_aggregate_order_by/case.toml", "tests/perf/query_cases/sql_join_filter_order/case.toml", diff --git a/config/config.md b/config/config.md index 86374927dd..7497842312 100644 --- a/config/config.md +++ b/config/config.md @@ -33,7 +33,7 @@ | `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. | | `http.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. | | `http.experimental_enable_prometheus_native_histogram` | Bool | `false` | Experimental: enable Prometheus remote write v2 native histogram ingestion. | -| `http.experimental_enable_explain_analyze_stream` | Bool | `false` | Experimental: enable POST /v1/sql/analyze/stream for streaming EXPLAIN ANALYZE VERBOSE metrics. | +| `http.experimental_enable_explain_analyze_stream` | Bool | `true` | Experimental: enable POST /v1/sql/analyze/stream for streaming EXPLAIN ANALYZE VERBOSE metrics. | | `grpc` | -- | -- | The gRPC server options. | | `grpc.bind_addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. | | `grpc.runtime_size` | Integer | `8` | The number of server worker threads. | @@ -253,7 +253,7 @@ | `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. | | `http.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. | | `http.experimental_enable_prometheus_native_histogram` | Bool | `false` | Experimental: enable Prometheus remote write v2 native histogram ingestion. | -| `http.experimental_enable_explain_analyze_stream` | Bool | `false` | Experimental: enable POST /v1/sql/analyze/stream for streaming EXPLAIN ANALYZE VERBOSE metrics. | +| `http.experimental_enable_explain_analyze_stream` | Bool | `true` | Experimental: enable POST /v1/sql/analyze/stream for streaming EXPLAIN ANALYZE VERBOSE metrics. | | `grpc` | -- | -- | The gRPC server options. | | `grpc.bind_addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. | | `grpc.server_addr` | String | `127.0.0.1:4001` | The address advertised to the metasrv, and used for connections from outside the host.
If left empty or unset, the server will automatically use the IP address of the first network interface
on the host, with the same port number as the one specified in `grpc.bind_addr`. | diff --git a/config/frontend.example.toml b/config/frontend.example.toml index 73efe8c137..3deadcf6a4 100644 --- a/config/frontend.example.toml +++ b/config/frontend.example.toml @@ -68,7 +68,7 @@ prom_validation_mode = "strict" ## Experimental: enable Prometheus remote write v2 native histogram ingestion. experimental_enable_prometheus_native_histogram = false ## Experimental: enable POST /v1/sql/analyze/stream for streaming EXPLAIN ANALYZE VERBOSE metrics. -experimental_enable_explain_analyze_stream = false +experimental_enable_explain_analyze_stream = true ## The gRPC server options. [grpc] diff --git a/config/standalone.example.toml b/config/standalone.example.toml index 571353ec20..1d612a2ff6 100644 --- a/config/standalone.example.toml +++ b/config/standalone.example.toml @@ -82,7 +82,7 @@ prom_validation_mode = "strict" ## Experimental: enable Prometheus remote write v2 native histogram ingestion. experimental_enable_prometheus_native_histogram = false ## Experimental: enable POST /v1/sql/analyze/stream for streaming EXPLAIN ANALYZE VERBOSE metrics. -experimental_enable_explain_analyze_stream = false +experimental_enable_explain_analyze_stream = true ## The gRPC server options. [grpc] diff --git a/src/client/src/region.rs b/src/client/src/region.rs index 983e2fd7b8..f1730a93c8 100644 --- a/src/client/src/region.rs +++ b/src/client/src/region.rs @@ -35,6 +35,7 @@ use common_recordbatch::{RecordBatch, RecordBatchStreamWrapper, SendableRecordBa use common_telemetry::error; use common_telemetry::tracing::Span; use common_telemetry::tracing_context::TracingContext; +use futures_util::Stream; use prost::Message; use query::query_engine::DefaultSerializer; use snafu::{OptionExt, ResultExt, location}; @@ -129,121 +130,10 @@ impl RegionRequester { let flight_data_stream = response.into_inner(); let mut decoder = FlightDecoder::default(); - let mut flight_message_stream = flight_data_stream + let flight_message_stream = flight_data_stream .filter_map(move |flight_data| decode_flight_data(&mut decoder, flight_data)); - let Some(first_flight_message) = flight_message_stream.next().await else { - return IllegalFlightMessagesSnafu { - reason: "Expect the response not to be empty", - } - .fail(); - }; - let FlightMessage::Schema(schema) = first_flight_message? else { - return IllegalFlightMessagesSnafu { - reason: "Expect schema to be the first flight message", - } - .fail(); - }; - - let metrics = Arc::new(ArcSwapOption::from(None)); - let metrics_ref = metrics.clone(); - - let tracing_context = TracingContext::from_current_span(); - - let schema = Arc::new( - datatypes::schema::Schema::try_from(schema).context(error::ConvertSchemaSnafu)?, - ); - let schema_cloned = schema.clone(); - let stream = Box::pin(stream!({ - let _span = tracing_context.attach(common_telemetry::tracing::info_span!( - "poll_flight_data_stream" - )); - - let mut buffered_message: Option = None; - let mut stream_ended = false; - - while !stream_ended { - // get the next message from the buffered message or read from the flight message stream - let flight_message_item = if let Some(msg) = buffered_message.take() { - Some(Ok(msg)) - } else { - flight_message_stream.next().await - }; - - let flight_message = match flight_message_item { - Some(Ok(message)) => message, - Some(Err(e)) => { - yield Err(BoxedError::new(e)).context(ExternalSnafu); - break; - } - None => break, - }; - - match flight_message { - FlightMessage::RecordBatch(record_batch) => { - let result_to_yield = - RecordBatch::from_df_record_batch(schema_cloned.clone(), record_batch); - - // get the next message from the stream. normally it should be a metrics message. - if let Some(next_flight_message_result) = flight_message_stream.next().await - { - match next_flight_message_result { - Ok(FlightMessage::Metrics(s)) => { - let m = serde_json::from_str(&s).ok().map(Arc::new); - metrics_ref.swap(m); - } - Ok(FlightMessage::RecordBatch(rb)) => { - // for some reason it's not a metrics message, so we need to buffer this record batch - // and yield it in the next iteration. - buffered_message = Some(FlightMessage::RecordBatch(rb)); - } - Ok(_) => { - yield IllegalFlightMessagesSnafu { - reason: "A RecordBatch message can only be succeeded by a Metrics message or another RecordBatch message" - } - .fail() - .map_err(BoxedError::new) - .context(ExternalSnafu); - break; - } - Err(e) => { - yield Err(BoxedError::new(e)).context(ExternalSnafu); - break; - } - } - } else { - // the stream has ended - stream_ended = true; - } - - yield Ok(result_to_yield); - } - FlightMessage::Metrics(s) => { - // just a branch in case of some metrics message comes after other things. - let m = serde_json::from_str(&s).ok().map(Arc::new); - metrics_ref.swap(m); - break; - } - _ => { - yield IllegalFlightMessagesSnafu { - reason: "A Schema message must be succeeded exclusively by a set of RecordBatch messages" - } - .fail() - .map_err(BoxedError::new) - .context(ExternalSnafu); - break; - } - } - } - })); - let record_batch_stream = RecordBatchStreamWrapper { - schema, - stream, - output_ordering: None, - metrics, - span: Span::current(), - }; - Ok(Box::pin(record_batch_stream)) + recordbatches_from_flight_message_stream(flight_message_stream).await } async fn handle_inner(&self, request: RegionRequest) -> Result { @@ -304,6 +194,124 @@ impl RegionRequester { } } +async fn recordbatches_from_flight_message_stream( + mut flight_message_stream: S, +) -> Result +where + S: Stream> + Send + Unpin + 'static, +{ + let Some(first_flight_message) = flight_message_stream.next().await else { + return IllegalFlightMessagesSnafu { + reason: "Expect the response not to be empty", + } + .fail(); + }; + let FlightMessage::Schema(schema) = first_flight_message? else { + return IllegalFlightMessagesSnafu { + reason: "Expect schema to be the first flight message", + } + .fail(); + }; + + let metrics = Arc::new(ArcSwapOption::from(None)); + let metrics_ref = metrics.clone(); + + let tracing_context = TracingContext::from_current_span(); + + let schema = + Arc::new(datatypes::schema::Schema::try_from(schema).context(error::ConvertSchemaSnafu)?); + let schema_cloned = schema.clone(); + let stream = Box::pin(stream!({ + let _span = tracing_context.attach(common_telemetry::tracing::info_span!( + "poll_flight_data_stream" + )); + + let mut buffered_message: Option = None; + let mut stream_ended = false; + + while !stream_ended { + // get the next message from the buffered message or read from the flight message stream + let flight_message_item = if let Some(msg) = buffered_message.take() { + Some(Ok(msg)) + } else { + flight_message_stream.next().await + }; + + let flight_message = match flight_message_item { + Some(Ok(message)) => message, + Some(Err(e)) => { + yield Err(BoxedError::new(e)).context(ExternalSnafu); + break; + } + None => break, + }; + + match flight_message { + FlightMessage::RecordBatch(record_batch) => { + let result_to_yield = + RecordBatch::from_df_record_batch(schema_cloned.clone(), record_batch); + + // get the next message from the stream. normally it should be a metrics message. + if let Some(next_flight_message_result) = flight_message_stream.next().await { + match next_flight_message_result { + Ok(FlightMessage::Metrics(s)) => { + let m = serde_json::from_str(&s).ok().map(Arc::new); + metrics_ref.swap(m); + } + Ok(FlightMessage::RecordBatch(rb)) => { + // for some reason it's not a metrics message, so we need to buffer this record batch + // and yield it in the next iteration. + buffered_message = Some(FlightMessage::RecordBatch(rb)); + } + Ok(_) => { + yield IllegalFlightMessagesSnafu { + reason: "A RecordBatch message can only be succeeded by a Metrics message or another RecordBatch message" + } + .fail() + .map_err(BoxedError::new) + .context(ExternalSnafu); + break; + } + Err(e) => { + yield Err(BoxedError::new(e)).context(ExternalSnafu); + break; + } + } + } else { + // the stream has ended + stream_ended = true; + } + + yield Ok(result_to_yield); + } + FlightMessage::Metrics(s) => { + // just a branch in case of some metrics message comes after other things. + let m = serde_json::from_str(&s).ok().map(Arc::new); + metrics_ref.swap(m); + continue; + } + _ => { + yield IllegalFlightMessagesSnafu { + reason: "A Schema message must be succeeded exclusively by a set of RecordBatch messages" + } + .fail() + .map_err(BoxedError::new) + .context(ExternalSnafu); + break; + } + } + } + })); + let record_batch_stream = RecordBatchStreamWrapper { + schema, + stream, + output_ordering: None, + metrics, + span: Span::current(), + }; + Ok(Box::pin(record_batch_stream)) +} + pub fn build_remote_dyn_filter_update_request( query_id: impl Into, update: RemoteDynFilterUpdate, @@ -371,9 +379,31 @@ mod test { use api::v1::region::{ RemoteDynFilterUnregister, RemoteDynFilterUpdate, region_request, remote_dyn_filter_request, }; + use common_recordbatch::adapter::RecordBatchMetrics; + use datatypes::prelude::{ConcreteDataType, VectorRef}; + use datatypes::schema::{ColumnSchema, Schema}; + use datatypes::vectors::Int32Vector; + use futures_util::stream; + use tonic::Status; use super::*; - use crate::Error::{IllegalDatabaseResponse, Server}; + use crate::Error::{self, IllegalDatabaseResponse, Server}; + + fn test_schema() -> Arc { + Arc::new(Schema::new(vec![ColumnSchema::new( + "v", + ConcreteDataType::int32_datatype(), + false, + )])) + } + + fn test_metrics_json() -> String { + serde_json::to_string(&RecordBatchMetrics { + elapsed_compute: 7, + ..Default::default() + }) + .unwrap() + } #[test] fn test_check_response_header() { @@ -469,4 +499,49 @@ mod test { Some(remote_dyn_filter_request::Action::Unregister(_)) )); } + + #[tokio::test] + async fn test_record_batch_stream_continues_after_pre_batch_metrics() { + let schema = test_schema(); + let batch = RecordBatch::new( + schema.clone(), + vec![Arc::new(Int32Vector::from_slice([1])) as VectorRef], + ) + .unwrap(); + + let mut recordbatches = recordbatches_from_flight_message_stream(stream::iter(vec![ + Ok(FlightMessage::Schema(schema.arrow_schema().clone())), + Ok(FlightMessage::Metrics(test_metrics_json())), + Ok(FlightMessage::RecordBatch(batch.into_df_record_batch())), + ])) + .await + .unwrap(); + + let batch = recordbatches.next().await.unwrap().unwrap(); + assert_eq!(batch.num_rows(), 1); + assert!(recordbatches.next().await.is_none()); + + let metrics = recordbatches.metrics().unwrap(); + assert_eq!(metrics.elapsed_compute, 7); + } + + #[tokio::test] + async fn test_record_batch_stream_exposes_error_after_pre_batch_metrics() { + let schema = test_schema(); + let mut recordbatches = recordbatches_from_flight_message_stream(stream::iter(vec![ + Ok(FlightMessage::Schema(schema.arrow_schema().clone())), + Ok(FlightMessage::Metrics(test_metrics_json())), + Err(Error::from(Status::internal("boom after metrics"))), + ])) + .await + .unwrap(); + + let err = recordbatches.next().await.unwrap().unwrap_err(); + assert_eq!("External error", err.to_string()); + assert!( + format!("{err:?}").contains("boom after metrics"), + "unexpected error: {err:?}" + ); + assert!(recordbatches.next().await.is_none()); + } } diff --git a/src/cmd/src/bin/query_perf_fixture/case.rs b/src/cmd/src/bin/query_perf_fixture/case.rs index 7dce735d7b..5bde9c83bf 100644 --- a/src/cmd/src/bin/query_perf_fixture/case.rs +++ b/src/cmd/src/bin/query_perf_fixture/case.rs @@ -36,6 +36,8 @@ pub(super) enum Scenario { pub(super) struct DirectReadableSstScenario { #[serde(default)] pub(super) seed: Option, + #[serde(default)] + pub(super) queries: Vec, pub(super) tables: Vec, pub(super) layout: LayoutConfig, } diff --git a/src/common/recordbatch/src/adapter.rs b/src/common/recordbatch/src/adapter.rs index 5529f0faf8..eeac94904b 100644 --- a/src/common/recordbatch/src/adapter.rs +++ b/src/common/recordbatch/src/adapter.rs @@ -286,14 +286,16 @@ impl RecordBatchStreamAdapter { }; match &self.metrics_2 { - Metrics::Unresolved(df_plan) | Metrics::PartialResolved(df_plan, _) => { + Metrics::Unresolved(df_plan) => { let metrics = collect_lightweight_query_load_metrics( df_plan.as_ref(), self.query_load_region_id, ); record_query_stats(counters, &metrics); } - Metrics::Resolved(metrics) => record_query_stats(counters, metrics), + Metrics::PartialResolved(_, metrics) | Metrics::Resolved(metrics) => { + record_query_stats(counters, metrics); + } Metrics::Unavailable => {} } } @@ -310,6 +312,61 @@ impl RecordBatchStreamAdapter { pub fn set_explain_verbose(&mut self, verbose: bool) { self.explain_verbose = verbose; } + + fn collect_plan_metrics(&self, df_plan: &Arc) -> RecordBatchMetrics { + collect_full_metrics( + df_plan.as_ref(), + self.explain_verbose, + self.query_load_region_id, + ) + } + + fn collect_partial_metrics( + df_plan: &dyn ExecutionPlan, + explain_verbose: bool, + query_load_region_id: Option, + ) -> RecordBatchMetrics { + if explain_verbose { + collect_full_metrics(df_plan, explain_verbose, query_load_region_id) + } else { + collect_lightweight_query_load_metrics(df_plan, query_load_region_id) + } + } + + fn update_plan_metrics(&mut self, final_metrics: bool) { + if final_metrics { + let df_plan = match &self.metrics_2 { + Metrics::Unresolved(df_plan) | Metrics::PartialResolved(df_plan, _) => { + df_plan.clone() + } + Metrics::Unavailable | Metrics::Resolved(_) => return, + }; + let metrics = self.collect_plan_metrics(&df_plan); + self.metrics_2 = Metrics::Resolved(metrics); + } else { + let explain_verbose = self.explain_verbose; + let query_load_region_id = self.query_load_region_id; + match &mut self.metrics_2 { + Metrics::Unresolved(df_plan) => { + let df_plan = df_plan.clone(); + let metrics = Self::collect_partial_metrics( + df_plan.as_ref(), + explain_verbose, + query_load_region_id, + ); + self.metrics_2 = Metrics::PartialResolved(df_plan, metrics); + } + Metrics::PartialResolved(df_plan, metrics) => { + *metrics = Self::collect_partial_metrics( + df_plan.as_ref(), + explain_verbose, + query_load_region_id, + ); + } + Metrics::Unavailable | Metrics::Resolved(_) => {} + } + } + } } /// Extracts total `output_bytes` from region scan plan nodes. @@ -420,10 +477,20 @@ impl RecordBatchStream for RecordBatchStreamAdapter { fn metrics(&self) -> Option { match &self.metrics_2 { - Metrics::Resolved(metrics) | Metrics::PartialResolved(_, metrics) => { - Some(metrics.clone()) + Metrics::Unresolved(df_plan) => { + if self.explain_verbose { + Some(self.collect_plan_metrics(df_plan)) + } else { + None + } } - Metrics::Unavailable | Metrics::Unresolved(_) => None, + Metrics::PartialResolved(df_plan, metrics) => Some(if self.explain_verbose { + self.collect_plan_metrics(df_plan) + } else { + metrics.clone() + }), + Metrics::Resolved(metrics) => Some(metrics.clone()), + Metrics::Unavailable => None, } } @@ -448,47 +515,14 @@ impl Stream for RecordBatchStreamAdapter { Poll::Pending => Poll::Pending, Poll::Ready(Some(df_record_batch)) => { let df_record_batch = df_record_batch?; - // Verbose analyze streams need complete partial metrics. Normal - // queries only need query-load metrics before EOF so early - // stop/cancellation can still report per-region load. - if self.explain_verbose { - if let Metrics::Unresolved(df_plan) | Metrics::PartialResolved(df_plan, _) = - &self.metrics_2 - { - let record_batch_metrics = collect_full_metrics( - df_plan.as_ref(), - self.explain_verbose, - self.query_load_region_id, - ); - self.metrics_2 = - Metrics::PartialResolved(df_plan.clone(), record_batch_metrics); - } - } else if let Metrics::Unresolved(df_plan) | Metrics::PartialResolved(df_plan, _) = - &self.metrics_2 - { - let record_batch_metrics = collect_lightweight_query_load_metrics( - df_plan.as_ref(), - self.query_load_region_id, - ); - self.metrics_2 = - Metrics::PartialResolved(df_plan.clone(), record_batch_metrics); - } + self.update_plan_metrics(false); Poll::Ready(Some(Ok(RecordBatch::from_df_record_batch( self.schema(), df_record_batch, )))) } Poll::Ready(None) => { - if let Metrics::Unresolved(df_plan) | Metrics::PartialResolved(df_plan, _) = - &self.metrics_2 - { - let record_batch_metrics = collect_full_metrics( - df_plan.as_ref(), - self.explain_verbose, - self.query_load_region_id, - ); - self.metrics_2 = Metrics::Resolved(record_batch_metrics); - } + self.update_plan_metrics(true); Poll::Ready(None) } } @@ -1060,7 +1094,7 @@ mod test { } #[test] - fn test_record_batch_stream_adapter_refreshes_partial_query_stats_on_drop() { + fn test_record_batch_stream_adapter_reuses_partial_query_stats_on_drop() { let schema = Arc::new(Schema::new(vec![ColumnSchema::new( "a", ConcreteDataType::int32_datatype(), @@ -1102,8 +1136,8 @@ mod test { drop(adapter); - assert_eq!(counters.query_cpu_time.load(Ordering::Relaxed), 52); - assert_eq!(counters.query_scanned_bytes.load(Ordering::Relaxed), 44); + assert_eq!(counters.query_cpu_time.load(Ordering::Relaxed), 11); + assert_eq!(counters.query_scanned_bytes.load(Ordering::Relaxed), 22); } #[tokio::test] diff --git a/src/datanode/src/region_server.rs b/src/datanode/src/region_server.rs index 5a6ab70085..367127df16 100644 --- a/src/datanode/src/region_server.rs +++ b/src/datanode/src/region_server.rs @@ -68,7 +68,9 @@ use servers::error::{ use servers::grpc::FlightCompression; use servers::grpc::flight::{FlightCraft, FlightRecordBatchStream, TonicStream}; use servers::grpc::region_server::RegionServerHandler; -use session::context::{QueryContext, QueryContextBuilder, QueryContextRef}; +use session::context::{ + FLIGHT_METRICS_HEARTBEAT_INTERVAL, QueryContext, QueryContextBuilder, QueryContextRef, +}; use snafu::{OptionExt, ResultExt, ensure}; use store_api::metric_engine_consts::{ FILE_ENGINE_NAME, LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME, @@ -84,7 +86,7 @@ use store_api::region_request::{ }; use store_api::storage::RegionId; use tokio::sync::{OwnedSemaphorePermit, Semaphore, mpsc}; -use tokio::time::timeout; +use tokio::time::{self, timeout}; use tonic::{Request, Response, Result as TonicResult}; use crate::error::{ @@ -1803,6 +1805,7 @@ impl RegionServerInner { request: QueryRequest, query_ctx: QueryContextRef, ) -> Result { + let explain_verbose = query_ctx.explain_verbose(); let inner = self.clone(); let mut stream = common_runtime::spawn_query(async move { inner.handle_read_inner(request, query_ctx).await @@ -1819,10 +1822,27 @@ impl RegionServerInner { let producer_metrics = metrics.clone(); let producer_handle = common_runtime::spawn_query(async move { - while let Some(batch) = stream.next().await { - *producer_metrics.write().unwrap() = stream.metrics(); - if sender.send(batch).await.is_err() { - break; + if explain_verbose { + loop { + match time::timeout(FLIGHT_METRICS_HEARTBEAT_INTERVAL, stream.next()).await { + Ok(Some(batch)) => { + *producer_metrics.write().unwrap() = stream.metrics(); + if sender.send(batch).await.is_err() { + return; + } + } + Ok(None) => break, + Err(_) => { + *producer_metrics.write().unwrap() = stream.metrics(); + } + } + } + } else { + while let Some(batch) = stream.next().await { + *producer_metrics.write().unwrap() = stream.metrics(); + if sender.send(batch).await.is_err() { + break; + } } } *producer_metrics.write().unwrap() = stream.metrics(); diff --git a/src/query/src/dist_plan/merge_scan.rs b/src/query/src/dist_plan/merge_scan.rs index 126ee9cb06..c05a17cb21 100644 --- a/src/query/src/dist_plan/merge_scan.rs +++ b/src/query/src/dist_plan/merge_scan.rs @@ -45,10 +45,14 @@ use futures_util::StreamExt; use greptime_proto::v1::region::RegionRequestHeader; use meter_core::data::ReadItem; use meter_macros::read_meter; -use session::context::QueryContextRef; +use session::context::{ + FLIGHT_METRICS_HEARTBEAT_INTERVAL, QueryContextRef, + SUPPORT_FLIGHT_METRICS_BEFORE_BATCH_EXTENSION_KEY, +}; use store_api::metrics::{REGION_QUERY_CPU_TIME, REGION_QUERY_SCANNED_BYTES}; use store_api::storage::RegionId; use table::table_name::TableName; +use tokio::time; use tokio::time::Instant; use tracing::{Instrument, Span}; @@ -388,12 +392,21 @@ impl MergeScanExec { region_id = %region_id, partition = partition )); - let region_query_ctx = query_context_for_remote_dyn_filter_region( + let mut region_query_ctx = query_context_for_remote_dyn_filter_region( &query_ctx, region_id, remote_dyn_filter_registry_lease.as_ref(), &captured_remote_dyn_filters, ); + if explain_verbose { + let remote_query_id = region_query_ctx.remote_query_id().map(str::to_string); + if let Some(remote_query_id) = remote_query_id { + region_query_ctx.set_extension( + SUPPORT_FLIGHT_METRICS_BEFORE_BATCH_EXTENSION_KEY, + remote_query_id, + ); + } + } let request = QueryRequest { header: Some(RegionRequestHeader { tracing_context: tracing_context.to_w3c(), @@ -438,7 +451,30 @@ impl MergeScanExec { let mut poll_duration = Duration::ZERO; let mut poll_timer = Instant::now(); - while let Some(batch) = stream.next().instrument(region_span.clone()).await { + loop { + let batch = if explain_verbose { + match time::timeout( + FLIGHT_METRICS_HEARTBEAT_INTERVAL, + stream.next().instrument(region_span.clone()), + ) + .await + { + Ok(batch) => batch, + Err(_) => { + if let Some(metrics) = stream.metrics() { + let mut sub_stage_metrics = + sub_stage_metrics_moved.lock().unwrap(); + sub_stage_metrics.insert(region_id, metrics); + } + continue; + } + } + } else { + stream.next().instrument(region_span.clone()).await + }; + let Some(batch) = batch else { + break; + }; let poll_elapsed = poll_timer.elapsed(); poll_duration += poll_elapsed; diff --git a/src/servers/src/grpc/flight/stream.rs b/src/servers/src/grpc/flight/stream.rs index ee08f8f20d..22dcd7baba 100644 --- a/src/servers/src/grpc/flight/stream.rs +++ b/src/servers/src/grpc/flight/stream.rs @@ -28,9 +28,13 @@ use futures::channel::mpsc; use futures::channel::mpsc::Sender; use futures::{SinkExt, Stream, StreamExt}; use pin_project::{pin_project, pinned_drop}; -use session::context::QueryContextRef; +use session::context::{ + FLIGHT_METRICS_HEARTBEAT_INTERVAL, QueryContextRef, + SUPPORT_FLIGHT_METRICS_BEFORE_BATCH_EXTENSION_KEY, +}; use snafu::ResultExt; use tokio::task::JoinHandle; +use tokio::time; use crate::error; use crate::grpc::FlightCompression; @@ -102,6 +106,35 @@ pub struct FlightRecordBatchStream { } impl FlightRecordBatchStream { + async fn send_metrics( + tx: &mut Sender>, + metrics: &mut StreamMetrics, + metrics_str: String, + ) -> bool { + metrics.metrics_count += 1; + let start = Instant::now(); + if let Err(e) = tx.send(Ok(FlightMessage::Metrics(metrics_str))).await { + warn!(e; "stop sending Flight data"); + return false; + } + metrics.send_metrics_duration += start.elapsed(); + true + } + + async fn send_metrics_if_changed( + tx: &mut Sender>, + metrics: &mut StreamMetrics, + last_metrics_str: &mut Option, + metrics_str: String, + ) -> bool { + if last_metrics_str.as_deref() == Some(metrics_str.as_str()) { + return true; + } + + *last_metrics_str = Some(metrics_str.clone()); + Self::send_metrics(tx, metrics, metrics_str).await + } + pub fn new( recordbatches: SendableRecordBatchStream, tracing_context: TracingContext, @@ -109,11 +142,20 @@ impl FlightRecordBatchStream { query_ctx: QueryContextRef, ) -> Self { let should_send_partial_metrics = query_ctx.explain_verbose(); + let can_send_metrics_before_batch = query_ctx + .remote_query_id() + .zip(query_ctx.extension(SUPPORT_FLIGHT_METRICS_BEFORE_BATCH_EXTENSION_KEY)) + .is_some_and(|(remote_query_id, capability)| capability == remote_query_id); let (tx, rx) = mpsc::channel::>(1); let join_handle = common_runtime::spawn_global(async move { - Self::flight_data_stream(recordbatches, tx, should_send_partial_metrics) - .trace(tracing_context.attach(info_span!("flight_data_stream"))) - .await + Self::flight_data_stream( + recordbatches, + tx, + should_send_partial_metrics, + can_send_metrics_before_batch, + ) + .trace(tracing_context.attach(info_span!("flight_data_stream"))) + .await }); let encoder = if compression.arrow_compression() { FlightEncoder::default() @@ -133,8 +175,10 @@ impl FlightRecordBatchStream { mut recordbatches: SendableRecordBatchStream, mut tx: Sender>, should_send_partial_metrics: bool, + can_send_metrics_before_batch: bool, ) { let mut metrics = StreamMetrics::new(should_send_partial_metrics); + let mut last_metrics_str = None; let schema = recordbatches.schema().arrow_schema().clone(); let start = Instant::now(); @@ -144,12 +188,41 @@ impl FlightRecordBatchStream { } metrics.send_schema_duration += start.elapsed(); - while let Some(batch_or_err) = { + loop { let start = Instant::now(); - let result = recordbatches.next().in_current_span().await; + let batch_or_err = if should_send_partial_metrics && can_send_metrics_before_batch { + match time::timeout( + FLIGHT_METRICS_HEARTBEAT_INTERVAL, + recordbatches.next().in_current_span(), + ) + .await + { + Ok(result) => result, + Err(_) => { + if let Some(metrics_str) = recordbatches + .metrics() + .and_then(|m| serde_json::to_string(&m).ok()) + && !Self::send_metrics_if_changed( + &mut tx, + &mut metrics, + &mut last_metrics_str, + metrics_str, + ) + .await + { + return; + } + metrics.fetch_content_duration += start.elapsed(); + continue; + } + } + } else { + recordbatches.next().in_current_span().await + }; metrics.fetch_content_duration += start.elapsed(); - result - } { + let Some(batch_or_err) = batch_or_err else { + break; + }; match batch_or_err { Ok(recordbatch) => { metrics.total_rows += recordbatch.num_rows(); @@ -172,14 +245,12 @@ impl FlightRecordBatchStream { && let Some(metrics_str) = recordbatches .metrics() .and_then(|m| serde_json::to_string(&m).ok()) - { - metrics.metrics_count += 1; - let start = Instant::now(); - if let Err(e) = tx.send(Ok(FlightMessage::Metrics(metrics_str))).await { - warn!(e; "stop sending Flight data"); - return; + && { + last_metrics_str = Some(metrics_str.clone()); + !Self::send_metrics(&mut tx, &mut metrics, metrics_str).await } - metrics.send_metrics_duration += start.elapsed(); + { + return; } } Err(e) => { @@ -200,10 +271,7 @@ impl FlightRecordBatchStream { .metrics() .and_then(|m| serde_json::to_string(&m).ok()) { - metrics.metrics_count += 1; - let start = Instant::now(); - let _ = tx.send(Ok(FlightMessage::Metrics(metrics_str))).await; - metrics.send_metrics_duration += start.elapsed(); + let _ = Self::send_metrics(&mut tx, &mut metrics, metrics_str).await; } } } @@ -255,18 +323,100 @@ impl Stream for FlightRecordBatchStream { #[cfg(test)] mod test { + use std::pin::Pin; use std::sync::Arc; + use std::task::{Context, Poll}; + use std::time::Duration; use common_grpc::flight::{FlightDecoder, FlightMessage}; - use common_recordbatch::{RecordBatch, RecordBatches}; + use common_recordbatch::adapter::RecordBatchMetrics; + use common_recordbatch::{OrderOption, RecordBatch, RecordBatchStream, RecordBatches}; use datatypes::prelude::*; - use datatypes::schema::{ColumnSchema, Schema}; + use datatypes::schema::{ColumnSchema, Schema, SchemaRef}; use datatypes::vectors::Int32Vector; use futures::StreamExt; - use session::context::QueryContext; + use session::context::{QueryContext, SUPPORT_FLIGHT_METRICS_BEFORE_BATCH_EXTENSION_KEY}; use super::*; + struct PendingMetricsStream { + schema: SchemaRef, + metrics: RecordBatchMetrics, + } + + struct MetricsThenBatchStream { + schema: SchemaRef, + metrics: RecordBatchMetrics, + rx: tokio::sync::mpsc::UnboundedReceiver>, + } + + fn query_context_with_matching_capability() -> Arc { + let query_ctx = QueryContext::arc(); + let remote_query_id = query_ctx + .remote_query_id() + .expect("query context must have remote query id") + .to_string(); + let mut query_ctx = (*query_ctx).clone(); + query_ctx.set_extension( + SUPPORT_FLIGHT_METRICS_BEFORE_BATCH_EXTENSION_KEY, + remote_query_id, + ); + Arc::new(query_ctx) + } + + fn query_context_with_capability(capability: &str) -> Arc { + let mut query_ctx = (*QueryContext::arc()).clone(); + query_ctx.set_extension( + SUPPORT_FLIGHT_METRICS_BEFORE_BATCH_EXTENSION_KEY, + capability, + ); + Arc::new(query_ctx) + } + + impl RecordBatchStream for PendingMetricsStream { + fn schema(&self) -> SchemaRef { + self.schema.clone() + } + + fn output_ordering(&self) -> Option<&[OrderOption]> { + None + } + + fn metrics(&self) -> Option { + Some(self.metrics.clone()) + } + } + + impl Stream for PendingMetricsStream { + type Item = common_recordbatch::error::Result; + + fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Pending + } + } + + impl RecordBatchStream for MetricsThenBatchStream { + fn schema(&self) -> SchemaRef { + self.schema.clone() + } + + fn output_ordering(&self) -> Option<&[OrderOption]> { + None + } + + fn metrics(&self) -> Option { + Some(self.metrics.clone()) + } + } + + impl Stream for MetricsThenBatchStream { + type Item = common_recordbatch::error::Result; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.rx.poll_recv(cx) + } + } + #[tokio::test] async fn test_flight_record_batch_stream() { let schema = Arc::new(Schema::new(vec![ColumnSchema::new( @@ -315,4 +465,233 @@ mod test { _ => unreachable!(), } } + + #[tokio::test] + async fn test_flight_record_batch_stream_emits_metrics_while_pending() { + let schema = Arc::new(Schema::new(vec![ColumnSchema::new( + "a", + ConcreteDataType::int32_datatype(), + false, + )])); + let metrics = RecordBatchMetrics { + elapsed_compute: 42, + ..Default::default() + }; + let recordbatches = Box::pin(PendingMetricsStream { + schema: schema.clone(), + metrics, + }); + let query_ctx = query_context_with_matching_capability(); + query_ctx.set_explain_verbose(true); + let mut stream = FlightRecordBatchStream::new( + recordbatches, + TracingContext::default(), + FlightCompression::default(), + query_ctx, + ); + + let decoder = &mut FlightDecoder::default(); + let schema_data = stream.next().await.unwrap().unwrap(); + match decoder.try_decode(&schema_data).unwrap().unwrap() { + FlightMessage::Schema(actual_schema) => { + assert_eq!(&actual_schema, schema.arrow_schema()); + } + _ => unreachable!(), + } + + let metrics_data = tokio::time::timeout(Duration::from_secs(2), stream.next()) + .await + .unwrap() + .unwrap() + .unwrap(); + match decoder.try_decode(&metrics_data).unwrap().unwrap() { + FlightMessage::Metrics(metrics) => { + let metrics: RecordBatchMetrics = serde_json::from_str(&metrics).unwrap(); + assert_eq!(metrics.elapsed_compute, 42); + } + other => panic!("expected metrics message, got {other:?}"), + } + } + + #[tokio::test] + async fn test_flight_record_batch_stream_continues_after_pending_metrics() { + let schema = Arc::new(Schema::new(vec![ColumnSchema::new( + "a", + ConcreteDataType::int32_datatype(), + false, + )])); + let metrics = RecordBatchMetrics { + elapsed_compute: 42, + ..Default::default() + }; + let recordbatch = RecordBatch::new( + schema.clone(), + vec![Arc::new(Int32Vector::from_slice([1])) as VectorRef], + ) + .unwrap(); + let expected_recordbatch = recordbatch.df_record_batch().clone(); + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + let recordbatches = Box::pin(MetricsThenBatchStream { + schema: schema.clone(), + metrics, + rx, + }); + let query_ctx = query_context_with_matching_capability(); + query_ctx.set_explain_verbose(true); + let mut stream = FlightRecordBatchStream::new( + recordbatches, + TracingContext::default(), + FlightCompression::default(), + query_ctx, + ); + + let decoder = &mut FlightDecoder::default(); + let schema_data = stream.next().await.unwrap().unwrap(); + assert!(matches!( + decoder.try_decode(&schema_data).unwrap().unwrap(), + FlightMessage::Schema(_) + )); + + let metrics_data = tokio::time::timeout(Duration::from_secs(2), stream.next()) + .await + .unwrap() + .unwrap() + .unwrap(); + assert!(matches!( + decoder.try_decode(&metrics_data).unwrap().unwrap(), + FlightMessage::Metrics(_) + )); + + tx.send(Ok(recordbatch)).unwrap(); + let batch_data = tokio::time::timeout(Duration::from_secs(2), stream.next()) + .await + .unwrap() + .unwrap() + .unwrap(); + match decoder.try_decode(&batch_data).unwrap().unwrap() { + FlightMessage::RecordBatch(actual_recordbatch) => { + assert_eq!(&actual_recordbatch, &expected_recordbatch); + } + other => panic!("expected record batch after pending metrics, got {other:?}"), + } + } + + #[tokio::test] + async fn test_flight_record_batch_stream_requires_capability_for_pre_batch_metrics() { + let schema = Arc::new(Schema::new(vec![ColumnSchema::new( + "a", + ConcreteDataType::int32_datatype(), + false, + )])); + let recordbatches = Box::pin(PendingMetricsStream { + schema: schema.clone(), + metrics: RecordBatchMetrics { + elapsed_compute: 42, + ..Default::default() + }, + }); + let query_ctx = QueryContext::arc(); + query_ctx.set_explain_verbose(true); + let mut stream = FlightRecordBatchStream::new( + recordbatches, + TracingContext::default(), + FlightCompression::default(), + query_ctx, + ); + + let decoder = &mut FlightDecoder::default(); + let schema_data = stream.next().await.unwrap().unwrap(); + assert!(matches!( + decoder.try_decode(&schema_data).unwrap().unwrap(), + FlightMessage::Schema(_) + )); + assert!( + tokio::time::timeout( + FLIGHT_METRICS_HEARTBEAT_INTERVAL + Duration::from_millis(200), + stream.next() + ) + .await + .is_err(), + "pre-batch Metrics must be gated by client capability" + ); + } + + #[tokio::test] + async fn test_flight_record_batch_stream_rejects_spoofed_capability_for_pre_batch_metrics() { + let schema = Arc::new(Schema::new(vec![ColumnSchema::new( + "a", + ConcreteDataType::int32_datatype(), + false, + )])); + let recordbatches = Box::pin(PendingMetricsStream { + schema: schema.clone(), + metrics: RecordBatchMetrics { + elapsed_compute: 42, + ..Default::default() + }, + }); + let query_ctx = query_context_with_capability("true"); + query_ctx.set_explain_verbose(true); + let mut stream = FlightRecordBatchStream::new( + recordbatches, + TracingContext::default(), + FlightCompression::default(), + query_ctx, + ); + + let decoder = &mut FlightDecoder::default(); + let schema_data = stream.next().await.unwrap().unwrap(); + assert!(matches!( + decoder.try_decode(&schema_data).unwrap().unwrap(), + FlightMessage::Schema(_) + )); + assert!( + tokio::time::timeout( + FLIGHT_METRICS_HEARTBEAT_INTERVAL + Duration::from_millis(200), + stream.next() + ) + .await + .is_err(), + "pre-batch Metrics must reject a spoofed capability" + ); + } + + #[tokio::test] + async fn test_flight_record_batch_stream_requires_explain_verbose_for_pre_batch_metrics() { + let schema = Arc::new(Schema::new(vec![ColumnSchema::new( + "a", + ConcreteDataType::int32_datatype(), + false, + )])); + let recordbatches = Box::pin(PendingMetricsStream { + schema: schema.clone(), + metrics: RecordBatchMetrics { + elapsed_compute: 42, + ..Default::default() + }, + }); + let query_ctx = query_context_with_matching_capability(); + let mut stream = FlightRecordBatchStream::new( + recordbatches, + TracingContext::default(), + FlightCompression::default(), + query_ctx, + ); + + let decoder = &mut FlightDecoder::default(); + let schema_data = stream.next().await.unwrap().unwrap(); + assert!(matches!( + decoder.try_decode(&schema_data).unwrap().unwrap(), + FlightMessage::Schema(_) + )); + assert!( + tokio::time::timeout( + FLIGHT_METRICS_HEARTBEAT_INTERVAL + Duration::from_millis(200), + stream.next() + ) + .await + .is_err(), + "pre-batch Metrics must be gated by explain verbose even when capability is set" + ); + } } diff --git a/src/servers/src/http.rs b/src/servers/src/http.rs index e85be71e8a..10ffba39db 100644 --- a/src/servers/src/http.rs +++ b/src/servers/src/http.rs @@ -184,7 +184,7 @@ impl Default for HttpOptions { enable_cors: true, prom_validation_mode: PromValidationMode::Strict, experimental_enable_prometheus_native_histogram: false, - experimental_enable_explain_analyze_stream: false, + experimental_enable_explain_analyze_stream: true, } } } @@ -1455,7 +1455,11 @@ mod test { #[tokio::test] pub async fn test_analyze_stream_route_config_gate() { let (tx, _rx) = mpsc::channel(100); - let app = make_test_app_custom(tx, HttpOptions::default()); + let options = HttpOptions { + experimental_enable_explain_analyze_stream: false, + ..Default::default() + }; + let app = make_test_app_custom(tx, options); let client = TestClient::new(app).await; let res = client .post("/v1/sql/analyze/stream?sql=EXPLAIN%20ANALYZE%20VERBOSE%20SELECT%201") @@ -1464,11 +1468,7 @@ mod test { assert_eq!(res.status(), StatusCode::NOT_FOUND); let (tx, _rx) = mpsc::channel(100); - let options = HttpOptions { - experimental_enable_explain_analyze_stream: true, - ..Default::default() - }; - let app = make_test_app_custom(tx, options); + let app = make_test_app_custom(tx, HttpOptions::default()); let client = TestClient::new(app).await; let res = client .post("/v1/sql/analyze/stream?sql=EXPLAIN%20ANALYZE%20VERBOSE%20SELECT%201") diff --git a/src/session/src/context.rs b/src/session/src/context.rs index 3d22f812fd..8d5b3406f3 100644 --- a/src/session/src/context.rs +++ b/src/session/src/context.rs @@ -32,7 +32,9 @@ use datafusion_common::config::ConfigOptions; use derive_builder::Builder; use sql::dialect::{Dialect, GenericDialect, GreptimeDbDialect, MySqlDialect, PostgreSqlDialect}; -pub use crate::hints::REMOTE_QUERY_ID_EXTENSION_KEY; +pub use crate::hints::{ + REMOTE_QUERY_ID_EXTENSION_KEY, SUPPORT_FLIGHT_METRICS_BEFORE_BATCH_EXTENSION_KEY, +}; use crate::protocol_ctx::ProtocolCtx; use crate::query_id::QueryId; use crate::session_config::{PGByteaOutputValue, PGDateOrder, PGDateTimeStyle, PGIntervalStyle}; @@ -41,6 +43,8 @@ use crate::{MutableInner, ReadPreference}; pub type QueryContextRef = Arc; pub type ConnInfoRef = Arc; +pub const FLIGHT_METRICS_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(1); + const CURSOR_COUNT_WARNING_LIMIT: usize = 10; pub fn generate_remote_query_id() -> String { diff --git a/src/session/src/hints.rs b/src/session/src/hints.rs index c971de5967..35f5101e45 100644 --- a/src/session/src/hints.rs +++ b/src/session/src/hints.rs @@ -19,11 +19,14 @@ pub const HINTS_KEY_PREFIX: &str = "x-greptime-hint-"; pub const REMOTE_QUERY_ID_EXTENSION_KEY: &str = "remote_query_id"; pub const INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY: &str = "initial_remote_dyn_filter_registrations"; +pub const SUPPORT_FLIGHT_METRICS_BEFORE_BATCH_EXTENSION_KEY: &str = + "query.support_flight_metrics_before_batch"; pub const READ_PREFERENCE_HINT: &str = "read_preference"; -pub const RESERVED_EXTENSION_KEYS: [&str; 2] = [ +pub const RESERVED_EXTENSION_KEYS: [&str; 3] = [ REMOTE_QUERY_ID_EXTENSION_KEY, INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY, + SUPPORT_FLIGHT_METRICS_BEFORE_BATCH_EXTENSION_KEY, ]; /// Deprecated, use `HINTS_KEY` instead. @@ -51,6 +54,9 @@ mod tests { assert!(is_reserved_extension_key( INITIAL_REMOTE_DYN_FILTER_REGISTRATIONS_EXTENSION_KEY )); + assert!(is_reserved_extension_key( + SUPPORT_FLIGHT_METRICS_BEFORE_BATCH_EXTENSION_KEY + )); assert!(!is_reserved_extension_key(READ_PREFERENCE_HINT)); } } diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index f6af868239..d15d75900e 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -2083,7 +2083,7 @@ prom_validation_mode = "strict" experimental_enable_prometheus_native_histogram = false cors_allowed_origins = [] enable_cors = true -experimental_enable_explain_analyze_stream = false +experimental_enable_explain_analyze_stream = true [grpc] bind_addr = "127.0.0.1:4001" diff --git a/tests/compatibility/cases/analyze_verbose_remote_metrics_extension/case.toml b/tests/compatibility/cases/analyze_verbose_remote_metrics_extension/case.toml new file mode 100644 index 0000000000..eeaf9ccb88 --- /dev/null +++ b/tests/compatibility/cases/analyze_verbose_remote_metrics_extension/case.toml @@ -0,0 +1,8 @@ +name = "analyze_verbose_remote_metrics_extension" +reason = "Verify EXPLAIN ANALYZE VERBOSE remote scans continue to work after adding the internal Flight metrics capability extension." +introduced_by = "PR #8405" +topologies = ["distributed"] +from_range = ["*"] +to_range = ["*"] +features = ["query", "compatibility", "explain-analyze"] +owner = "query" diff --git a/tests/compatibility/cases/analyze_verbose_remote_metrics_extension/setup.sql b/tests/compatibility/cases/analyze_verbose_remote_metrics_extension/setup.sql new file mode 100644 index 0000000000..73e0788b70 --- /dev/null +++ b/tests/compatibility/cases/analyze_verbose_remote_metrics_extension/setup.sql @@ -0,0 +1,15 @@ +CREATE TABLE t_analyze_verbose_remote_metrics_extension( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY, + val INT +) +PARTITION ON COLUMNS (host) ( + host < 'm', + host >= 'm' +); + +INSERT INTO t_analyze_verbose_remote_metrics_extension VALUES +('2024-02-04 00:00:00+0000', 'host_a', 1), +('2024-02-04 00:01:00+0000', 'host_z', 2); + +ADMIN FLUSH_TABLE('t_analyze_verbose_remote_metrics_extension'); diff --git a/tests/compatibility/cases/analyze_verbose_remote_metrics_extension/verify.result b/tests/compatibility/cases/analyze_verbose_remote_metrics_extension/verify.result new file mode 100644 index 0000000000..39059c7d70 --- /dev/null +++ b/tests/compatibility/cases/analyze_verbose_remote_metrics_extension/verify.result @@ -0,0 +1,33 @@ +-- SQLNESS REPLACE (-+) - +-- SQLNESS REPLACE (\s\s+) _ +-- SQLNESS REPLACE (elapsed_compute.*) REDACTED +-- SQLNESS REPLACE (metrics.*) REDACTED +-- SQLNESS REPLACE (peers.*) REDACTED +-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED +-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED +-- SQLNESS REPLACE "partition_count":\{(.*?)\} "partition_count":REDACTED +-- SQLNESS REPLACE ("files":\s\[.*\],) "files": REDACTED, +-- SQLNESS REPLACE "flat_format":\s\w+, "flat_format": REDACTED, +EXPLAIN ANALYZE VERBOSE SELECT count(*) FROM t_analyze_verbose_remote_metrics_extension; + ++-+-+-+ +| stage | node | plan_| ++-+-+-+ +| 0_| 0_|_ProjectionExec: expr=[count(Int64(1))@0 as count(*)] REDACTED +|_|_|_AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))] REDACTED +|_|_|_CoalescePartitionsExec REDACTED +|_|_|_AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))] REDACTED +|_|_|_MergeScanExec: REDACTED +|_|_|_| +| 1_| 0_|_AggregateExec: mode=Final, gby=[], aggr=[__count_state(t_analyze_verbose_remote_REDACTED +|_|_|_CoalescePartitionsExec REDACTED +|_|_|_AggregateExec: mode=Partial, gby=[], aggr=[__count_state(t_analyze_verbose_remote_REDACTED +|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["ts"], "filters": ["host >= Utf8(\"m\") AND host IS NOT NULL"], "flat_format": REDACTED, "REDACTED +|_|_|_| +| 1_| 1_|_AggregateExec: mode=Final, gby=[], aggr=[__count_state(t_analyze_verbose_remote_REDACTED +|_|_|_CoalescePartitionsExec REDACTED +|_|_|_AggregateExec: mode=Partial, gby=[], aggr=[__count_state(t_analyze_verbose_remote_REDACTED +|_|_|_SeqScan: region=REDACTED, {"partition_count":REDACTED, "projection": ["ts"], "filters": ["host < Utf8(\"m\") OR host IS NULL"], "files": REDACTED, "flat_format": REDACTED, "REDACTED +|_|_|_| +|_|_| Total rows: 1_| ++-+-+-+ diff --git a/tests/compatibility/cases/analyze_verbose_remote_metrics_extension/verify.sql b/tests/compatibility/cases/analyze_verbose_remote_metrics_extension/verify.sql new file mode 100644 index 0000000000..81eb489e66 --- /dev/null +++ b/tests/compatibility/cases/analyze_verbose_remote_metrics_extension/verify.sql @@ -0,0 +1,11 @@ +-- SQLNESS REPLACE (-+) - +-- SQLNESS REPLACE (\s\s+) _ +-- SQLNESS REPLACE (elapsed_compute.*) REDACTED +-- SQLNESS REPLACE (metrics.*) REDACTED +-- SQLNESS REPLACE (peers.*) REDACTED +-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED +-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED +-- SQLNESS REPLACE "partition_count":\{(.*?)\} "partition_count":REDACTED +-- SQLNESS REPLACE ("files":\s\[.*\],) "files": REDACTED, +-- SQLNESS REPLACE "flat_format":\s\w+, "flat_format": REDACTED, +EXPLAIN ANALYZE VERBOSE SELECT count(*) FROM t_analyze_verbose_remote_metrics_extension; diff --git a/tests/perf/query_cases/analyze_verbose_many_files/case.toml b/tests/perf/query_cases/analyze_verbose_many_files/case.toml new file mode 100644 index 0000000000..bb9548fb73 --- /dev/null +++ b/tests/perf/query_cases/analyze_verbose_many_files/case.toml @@ -0,0 +1,80 @@ +# EXPLAIN ANALYZE VERBOSE over many files. +# +# This protects the remote analyze metrics heartbeat path from adding excessive +# overhead when verbose plan metrics are large. The query intentionally scans a +# huge direct-SST fixture so the candidate must handle very high file counts and +# the resulting verbose metrics payload without a large latency regression. + +[case] +name = "analyze_verbose_many_files" +description = "EXPLAIN ANALYZE VERBOSE should not regress significantly when scanning millions of SST files" + +[scenario] +kind = "direct_readable_sst" +seed = 8405 + + +[[scenario.tables]] +database = "public" +name = "analyze_verbose_many_files" +engine = "mito" +append_mode = true +sst_format = "flat" +primary_key = ["service", "host", "instance"] +time_index = "ts" + +[[scenario.tables.columns]] +name = "service" +type = "STRING" +semantic = "tag" +distribution = { kind = "cardinality", values = 16, prefix = "service" } + +[[scenario.tables.columns]] +name = "host" +type = "STRING" +semantic = "tag" +distribution = { kind = "cardinality", values = 256, prefix = "host" } + +[[scenario.tables.columns]] +name = "instance" +type = "STRING" +semantic = "tag" +distribution = { kind = "cardinality", values = 512, prefix = "instance" } + +[[scenario.tables.columns]] +name = "cpu" +type = "DOUBLE" +semantic = "field" +distribution = { kind = "deterministic_wave", min = 0.0, max = 100.0 } + +[[scenario.tables.columns]] +name = "mem" +type = "DOUBLE" +semantic = "field" +distribution = { kind = "deterministic_wave", min = 0.0, max = 4096.0 } + +[[scenario.tables.columns]] +name = "ts" +type = "TIMESTAMP(9)" +semantic = "timestamp" + +[scenario.layout] +regions = 1 +sst_count = 2000000 +rows_per_sst = 1 +row_group_size = 1 +series_count = 16 +start_unix_nanos = 1704067200000000000 # 2024-01-01T00:00:00Z +step_nanos = 1000000000 +time_range_layout = "non_overlapping_per_sst" +series_layout = "round_robin" + +[[scenario.queries]] +name = "explain_analyze_verbose_full_scan" +kind = "sql" +query = "EXPLAIN ANALYZE VERBOSE SELECT service, avg(cpu) AS avg_cpu, max(mem) AS max_mem FROM analyze_verbose_many_files GROUP BY service ORDER BY avg_cpu DESC LIMIT 10" +warmup = 1 +iterations = 3 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 30