rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-08-18 12:08:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: stream remote analyze metrics while pending (#8405)

Browse Source 
* fix: stream remote analyze metrics while pending

Signed-off-by: discord9 <discord9@163.com>

* test: verify flight metrics preserve pending batch

Signed-off-by: discord9 <discord9@163.com>

* fix: preserve direct SST perf queries in plans

Signed-off-by: discord9 <discord9@163.com>

* fix: bind flight metrics capability to query

Signed-off-by: discord9 <discord9@163.com>

---------

Signed-off-by: discord9 <discord9@163.com>
This commit is contained in:
discord9
2026-07-21 12:17:50 +08:00
committed by GitHub
parent e3d2864947
commit 56addd0623
19 changed files with 906 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/scripts/query-regression-run.py
+1
View File
@@ -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",
config/config.md
+2 -2
View File
@@ -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.<br/>Available options:<br/>- strict: deny invalid UTF-8 strings (default).<br/>- lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).<br/>- 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.<br/>Available options:<br/>- strict: deny invalid UTF-8 strings (default).<br/>- lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).<br/>- 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.<br/>If left empty or unset, the server will automatically use the IP address of the first network interface<br/>on the host, with the same port number as the one specified in `grpc.bind_addr`. |
config/frontend.example.toml
+1 -1
View File
@@ -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]
config/standalone.example.toml
+1 -1
View File
@@ -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]
src/client/src/region.rs
+189 -114
View File
@@ -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<FlightMessage> = 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<RegionResponse> {
@@ -304,6 +194,124 @@ impl RegionRequester {
}
}
async fn recordbatches_from_flight_message_stream<S>(
mut flight_message_stream: S,
) -> Result<SendableRecordBatchStream>
where
S: Stream<Item = Result<FlightMessage>> + 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<FlightMessage> = 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<String>,
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<Schema> {
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());
}
}
src/cmd/src/bin/query_perf_fixture/case.rs
+2
View File
@@ -36,6 +36,8 @@ pub(super) enum Scenario {
pub(super) struct DirectReadableSstScenario {
#[serde(default)]
pub(super) seed: Option<u64>,
#[serde(default)]
pub(super) queries: Vec<serde_json::Value>,
pub(super) tables: Vec<TableConfig>,
pub(super) layout: LayoutConfig,
}
src/common/recordbatch/src/adapter.rs
+77 -43
View File
@@ -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<dyn ExecutionPlan>) -> 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<u64>,
) -> 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<RecordBatchMetrics> {
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]
src/datanode/src/region_server.rs
+26 -6
View File
@@ -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<SendableRecordBatchStream> {
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();
src/query/src/dist_plan/merge_scan.rs
+39 -3
View File
@@ -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;
src/servers/src/grpc/flight/stream.rs
+401 -22
View File
@@ -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<TonicResult<FlightMessage>>,
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<TonicResult<FlightMessage>>,
metrics: &mut StreamMetrics,
last_metrics_str: &mut Option<String>,
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::<TonicResult<FlightMessage>>(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<TonicResult<FlightMessage>>,
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<common_recordbatch::error::Result<RecordBatch>>,
}
fn query_context_with_matching_capability() -> Arc<QueryContext> {
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<QueryContext> {
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<RecordBatchMetrics> {
Some(self.metrics.clone())
}
}
impl Stream for PendingMetricsStream {
type Item = common_recordbatch::error::Result<RecordBatch>;
fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
Poll::Pending
}
}
impl RecordBatchStream for MetricsThenBatchStream {
fn schema(&self) -> SchemaRef {
self.schema.clone()
}
fn output_ordering(&self) -> Option<&[OrderOption]> {
None
}
fn metrics(&self) -> Option<RecordBatchMetrics> {
Some(self.metrics.clone())
}
}
impl Stream for MetricsThenBatchStream {
type Item = common_recordbatch::error::Result<RecordBatch>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
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"
);
}
}
src/servers/src/http.rs
+7 -7
View File
@@ -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")
src/session/src/context.rs
+5 -1
View File
@@ -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<QueryContext>;
pub type ConnInfoRef = Arc<ConnInfo>;
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 {
src/session/src/hints.rs
+7 -1
View File
@@ -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));
}
}
tests-integration/tests/http.rs
+1 -1
View File
@@ -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"
tests/compatibility/cases/analyze_verbose_remote_metrics_extension/case.toml
+8
View File
@@ -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"
tests/compatibility/cases/analyze_verbose_remote_metrics_extension/setup.sql
+15
View File
@@ -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');
tests/compatibility/cases/analyze_verbose_remote_metrics_extension/verify.result
+33
View File
@@ -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_|
+-+-+-+
tests/compatibility/cases/analyze_verbose_remote_metrics_extension/verify.sql
+11
View File
@@ -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;
tests/perf/query_cases/analyze_verbose_many_files/case.toml
+80
View File
@@ -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