diff --git a/Cargo.lock b/Cargo.lock index a389391ea9..534f484d4f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -13470,6 +13470,7 @@ dependencies = [ "rustls", "rustls-pemfile", "rustls-pki-types", + "ryu", "serde", "serde_json", "servers", diff --git a/Cargo.toml b/Cargo.toml index e6b849a6cd..7c3fccfbbf 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -231,6 +231,7 @@ rstest_reuse = "0.7" rust_decimal = "1.33" rustc-hash = "2.0" rustls = { version = "0.23.25", default-features = false } +ryu = "1.0" sea-query = "0.32" serde = { version = "1.0", features = ["derive"] } serde_json = { version = "1.0", features = ["float_roundtrip"] } diff --git a/src/cmd/src/bin/query_regression_runner/direct.rs b/src/cmd/src/bin/query_regression_runner/direct.rs index 04c191c8c5..b003b6ee5d 100644 --- a/src/cmd/src/bin/query_regression_runner/direct.rs +++ b/src/cmd/src/bin/query_regression_runner/direct.rs @@ -166,7 +166,28 @@ fn create_table_sql(table: &Table) -> Result { let columns = table .columns .iter() - .map(|column| format!("{} {}", sql_ident(&column.name), column.data_type)) + .map(|column| { + // The direct-SST fixture generator (`query_perf_fixture::direct_sst:: + // build_region_metadata`) bakes index metadata into the region + // manifest's column schemas: tag columns get an inverted index and + // a skipping index with granularity 1, field columns get a skipping + // index with granularity 1. The CREATE TABLE must declare the same + // column options, otherwise the table schema (frontend) and the + // region schema (datanode, from the manifest) disagree and the + // MergeScan remote schema validation fails with "advertised remote + // stream schema field mismatch". + let mut sql = format!("{} {}", sql_ident(&column.name), column.data_type); + match column.semantic.as_deref() { + Some("tag") => { + sql.push_str(" SKIPPING INDEX WITH (granularity='1') INVERTED INDEX"); + } + Some("field") => { + sql.push_str(" SKIPPING INDEX WITH (granularity='1')"); + } + _ => {} + } + sql + }) .collect::>() .join(",\n "); let primary_key = table @@ -455,10 +476,12 @@ mod tests { Column { name: "host".to_string(), data_type: "STRING".to_string(), + semantic: Some("tag".to_string()), }, Column { name: "ts".to_string(), data_type: "TIMESTAMP(9)".to_string(), + semantic: Some("timestamp".to_string()), }, ], primary_key: vec!["host".to_string()], @@ -469,7 +492,86 @@ mod tests { }; assert_eq!( create_table_sql(&table).unwrap(), - "CREATE TABLE \"metric\"\"name\" (\n \"host\" STRING,\n \"ts\" TIMESTAMP(9),\n TIME INDEX (\"ts\"),\n PRIMARY KEY (\"host\")\n) ENGINE=mito\nWITH ('append_mode'='true', 'sst_format'='flat');" + "CREATE TABLE \"metric\"\"name\" (\n \"host\" STRING SKIPPING INDEX WITH (granularity='1') INVERTED INDEX,\n \"ts\" TIMESTAMP(9),\n TIME INDEX (\"ts\"),\n PRIMARY KEY (\"host\")\n) ENGINE=mito\nWITH ('append_mode'='true', 'sst_format'='flat');" + ); + } + + #[test] + fn create_table_sql_declares_indexes_matching_fixture_metadata() { + // Mirrors the index metadata baked by + // query_perf_fixture::direct_sst::build_region_metadata: tag columns + // get SKIPPING INDEX WITH (granularity='1') + INVERTED INDEX, field + // columns get SKIPPING INDEX WITH (granularity='1'), timestamp columns + // get no column options. Without these options the frontend table + // schema and the datanode region schema (from the fixture manifest) + // disagree and the MergeScan remote schema validation fails. + let table = Table { + database: "public".to_string(), + name: "metric".to_string(), + engine: "mito".to_string(), + columns: vec![ + Column { + name: "host".to_string(), + data_type: "STRING".to_string(), + semantic: Some("tag".to_string()), + }, + Column { + name: "instance".to_string(), + data_type: "STRING".to_string(), + semantic: Some("tag".to_string()), + }, + Column { + name: "value".to_string(), + data_type: "DOUBLE".to_string(), + semantic: Some("field".to_string()), + }, + Column { + name: "ts".to_string(), + data_type: "TIMESTAMP(9)".to_string(), + semantic: Some("timestamp".to_string()), + }, + ], + primary_key: vec!["host".to_string(), "instance".to_string()], + time_index: Some("ts".to_string()), + append_mode: Some(true), + sst_format: Some("flat".to_string()), + validate_show_create_engine: true, + }; + assert_eq!( + create_table_sql(&table).unwrap(), + "CREATE TABLE \"metric\" (\n \"host\" STRING SKIPPING INDEX WITH (granularity='1') INVERTED INDEX,\n \"instance\" STRING SKIPPING INDEX WITH (granularity='1') INVERTED INDEX,\n \"value\" DOUBLE SKIPPING INDEX WITH (granularity='1'),\n \"ts\" TIMESTAMP(9),\n TIME INDEX (\"ts\"),\n PRIMARY KEY (\"host\", \"instance\")\n) ENGINE=mito\nWITH ('append_mode'='true', 'sst_format'='flat');" + ); + } + + #[test] + fn create_table_sql_without_semantic_keeps_bare_columns() { + // Columns without a semantic type must keep the historical bare form so + // tables that predate the index-metadata fix are unaffected. + let table = Table { + database: "public".to_string(), + name: "metric".to_string(), + engine: "mito".to_string(), + columns: vec![ + Column { + name: "host".to_string(), + data_type: "STRING".to_string(), + semantic: None, + }, + Column { + name: "ts".to_string(), + data_type: "TIMESTAMP(9)".to_string(), + semantic: None, + }, + ], + primary_key: vec!["host".to_string()], + time_index: Some("ts".to_string()), + append_mode: None, + sst_format: None, + validate_show_create_engine: true, + }; + assert_eq!( + create_table_sql(&table).unwrap(), + "CREATE TABLE \"metric\" (\n \"host\" STRING,\n \"ts\" TIMESTAMP(9),\n TIME INDEX (\"ts\"),\n PRIMARY KEY (\"host\")\n) ENGINE=mito;" ); } diff --git a/src/cmd/src/bin/query_regression_runner/measure.rs b/src/cmd/src/bin/query_regression_runner/measure.rs index 2b389f7d78..fd80813f40 100644 --- a/src/cmd/src/bin/query_regression_runner/measure.rs +++ b/src/cmd/src/bin/query_regression_runner/measure.rs @@ -21,7 +21,7 @@ use serde_json::{Map, Value, json}; use crate::query_regression_runner::model::{Measurement, Query, QueryResult, Scenario, Table}; use crate::query_regression_runner::plan::{load_plan, normalize_scenario}; -use crate::query_regression_runner::sql::{http_post_sql, sql_ident}; +use crate::query_regression_runner::sql::{http_post_prom_range_query, http_post_sql, sql_ident}; use crate::query_regression_runner::{MeasureArgs, Result}; pub(super) async fn run_measure(args: MeasureArgs) -> Result<()> { @@ -106,6 +106,26 @@ fn target_report(name: &str, http_port: u16, result: QueryResult) -> Value { }) } +/// Routes a configured query to the right endpoint: `prom_http` queries hit +/// the Prometheus HTTP range API (which exercises the Prometheus JSON response +/// builder), everything else goes through `/v1/sql` (including `TQL ANALYZE`). +async fn post_query(client: &Client, port: u16, query: &Query, db: &str) -> Value { + if query.kind.as_deref() == Some("prom_http") { + http_post_prom_range_query( + client, + port, + &query.query, + query.start.as_deref(), + query.end.as_deref(), + query.step.as_deref(), + db, + ) + .await + } else { + http_post_sql(client, port, &query.query, db).await + } +} + async fn run_target( port: u16, tables: &[Table], @@ -118,6 +138,9 @@ async fn run_target( name: Some("count_all".to_string()), kind: Some("sql".to_string()), query: format!("SELECT count(*) FROM {}", sql_ident(&tables[0].name)), + start: None, + end: None, + step: None, warmup: 0, iterations: 1, thresholds: Map::new(), @@ -147,7 +170,7 @@ async fn run_target( } validation.push(sample); } - let first = http_post_sql(client, port, &queries[0].query, db).await; + let first = post_query(client, port, &queries[0], db).await; if !first["ok"].as_bool().unwrap_or(false) { validation_errors.push(json!({ "sql": queries[0].query, @@ -160,7 +183,7 @@ async fn run_target( let mut measurements = Vec::with_capacity(queries.len()); for query in &queries { for _ in 0..query.warmup { - let warmup = http_post_sql(client, port, &query.query, db).await; + let warmup = post_query(client, port, query, db).await; if !warmup["ok"].as_bool().unwrap_or(false) { validation_errors.push(json!({ "sql": query.query, @@ -173,10 +196,12 @@ async fn run_target( let mut samples = Vec::with_capacity(query.iterations); let mut good_latencies = Vec::with_capacity(query.iterations); for _ in 0..query.iterations { - let mut sample = http_post_sql(client, port, &query.query, db).await; + let mut sample = post_query(client, port, query, db).await; let execution_time = sample .get("response") - .and_then(extract_execution_time) + .and_then(|response| { + extract_execution_time_for_kind(query.kind.as_deref(), response) + }) .cloned() .unwrap_or(Value::Null); sample @@ -242,6 +267,14 @@ fn response_text(body: &Value) -> String { .unwrap_or_else(|| serde_json::to_string(body).unwrap_or_default()) } +fn extract_execution_time_for_kind<'a>(kind: Option<&str>, body: &'a Value) -> Option<&'a Value> { + if kind == Some("prom_http") { + None + } else { + extract_execution_time(body) + } +} + fn extract_execution_time(body: &Value) -> Option<&Value> { match body { Value::Object(map) => { @@ -379,12 +412,33 @@ mod tests { ); } + #[test] + fn prom_http_does_not_extract_execution_time_from_response() { + let response = json!({ + "data": { + "result": [{"metric": {"job": "api", "elapsed": "label"}}], + "elapsed": 42 + } + }); + assert_eq!( + extract_execution_time_for_kind(Some("prom_http"), &response), + None + ); + assert_eq!( + extract_execution_time_for_kind(Some("sql"), &response), + Some(&json!(42)) + ); + } + #[test] fn threshold_rejects_unknown_keys_and_zero_base() { let query = Query { name: Some("q".to_string()), kind: None, query: "SELECT 1".to_string(), + start: None, + end: None, + step: None, warmup: 0, iterations: 1, thresholds: Map::from_iter([ diff --git a/src/cmd/src/bin/query_regression_runner/model.rs b/src/cmd/src/bin/query_regression_runner/model.rs index 50d41b3f85..6831a5256a 100644 --- a/src/cmd/src/bin/query_regression_runner/model.rs +++ b/src/cmd/src/bin/query_regression_runner/model.rs @@ -90,6 +90,12 @@ pub(super) struct Column { pub(super) name: String, #[serde(rename = "type")] pub(super) data_type: String, + /// Semantic type ("tag", "field", "timestamp") from the case TOML. The + /// direct-SST fixture generator bakes index metadata into the region + /// manifest based on this semantic, so the CREATE TABLE must declare the + /// matching column options (see `create_table_sql` in `direct.rs`). + #[serde(default)] + pub(super) semantic: Option, } const fn default_show_create_engine() -> bool { @@ -203,6 +209,13 @@ pub(super) struct Query { #[serde(default)] pub(super) kind: Option, pub(super) query: String, + /// Prometheus HTTP range query parameters (`kind = "prom_http"` only). + #[serde(default)] + pub(super) start: Option, + #[serde(default)] + pub(super) end: Option, + #[serde(default)] + pub(super) step: Option, #[serde(default)] pub(super) warmup: usize, #[serde(default = "one")] diff --git a/src/cmd/src/bin/query_regression_runner/sql.rs b/src/cmd/src/bin/query_regression_runner/sql.rs index 4d8baab2c8..b33c0a1c22 100644 --- a/src/cmd/src/bin/query_regression_runner/sql.rs +++ b/src/cmd/src/bin/query_regression_runner/sql.rs @@ -117,10 +117,62 @@ pub(super) fn value_f64(value: Option<&Value>) -> Option { } pub(super) async fn http_post_sql(client: &Client, port: u16, sql: &str, db: &str) -> Value { + let mut sample = post_form( + client, + format!("http://127.0.0.1:{port}/v1/sql"), + &[("sql", sql), ("db", db), ("format", "json")], + ) + .await; + sample + .as_object_mut() + .expect("HTTP samples are objects") + .insert("sql".to_string(), Value::String(sql.to_string())); + sample +} + +/// Posts a Prometheus HTTP API range query (`/v1/prometheus/api/v1/query_range`) +/// and measures the full request-to-body latency. This exercises the Prometheus +/// JSON response building path (`PrometheusJsonResponse::record_batches_to_data`) +/// that `/v1/sql` (including `TQL ANALYZE`) does not go through. +pub(super) async fn http_post_prom_range_query( + client: &Client, + port: u16, + query: &str, + start: Option<&str>, + end: Option<&str>, + step: Option<&str>, + db: &str, +) -> Value { + let mut sample = post_form( + client, + prom_range_query_url(port, db), + &[ + ("query", query), + ("start", start.unwrap_or_default()), + ("end", end.unwrap_or_default()), + ("step", step.unwrap_or_default()), + ], + ) + .await; + sample + .as_object_mut() + .expect("HTTP samples are objects") + .insert("query".to_string(), Value::String(query.to_string())); + sample +} + +fn prom_range_query_url(port: u16, db: &str) -> String { + let mut url = reqwest::Url::parse(&format!( + "http://127.0.0.1:{port}/v1/prometheus/api/v1/query_range" + )) + .expect("fixed Prometheus range query URL must be valid"); + url.query_pairs_mut().append_pair("db", db); + url.into() +} + +async fn post_form(client: &Client, url: String, form: &[(&str, &str)]) -> Value { let started = Instant::now(); - let request = client - .post(format!("http://127.0.0.1:{port}/v1/sql")) - .form(&[("sql", sql), ("db", db), ("format", "json")]); + let request = client.post(url).form(form); match request.send().await { Ok(response) => { let status = response.status().as_u16(); @@ -133,7 +185,6 @@ pub(super) async fn http_post_sql(client: &Client, port: u16, sql: &str, db: &st "status": status, "latency_ms": started.elapsed().as_secs_f64() * 1000.0, "response": body, - "sql": sql, }); if status >= 400 { sample @@ -148,7 +199,6 @@ pub(super) async fn http_post_sql(client: &Client, port: u16, sql: &str, db: &st "status": status, "latency_ms": started.elapsed().as_secs_f64() * 1000.0, "error": error.to_string(), - "sql": sql, }), } } @@ -157,7 +207,6 @@ pub(super) async fn http_post_sql(client: &Client, port: u16, sql: &str, db: &st "status": Value::Null, "latency_ms": started.elapsed().as_secs_f64() * 1000.0, "error": error.to_string(), - "sql": sql, }), } } @@ -208,6 +257,17 @@ pub(super) fn sql_ident(name: &str) -> String { mod tests { use super::*; + #[test] + fn prom_range_query_url_places_database_in_query_parameters() { + let url = reqwest::Url::parse(&prom_range_query_url(4000, "catalog-schema name")) + .expect("generated URL must be valid"); + assert_eq!(url.path(), "/v1/prometheus/api/v1/query_range"); + assert_eq!( + url.query_pairs().collect::>(), + vec![("db".into(), "catalog-schema name".into())] + ); + } + #[test] fn top_level_errors_do_not_inspect_rows() { assert!(response_has_error(&json!({"error_code": 7}))); diff --git a/src/servers/Cargo.toml b/src/servers/Cargo.toml index 59d00b90ad..7936c96fbd 100644 --- a/src/servers/Cargo.toml +++ b/src/servers/Cargo.toml @@ -115,6 +115,7 @@ rust_decimal = { workspace = true, features = ["db-postgres"] } rustls = { workspace = true, default-features = false, features = ["aws_lc_rs", "logging", "std", "tls12"] } rustls-pemfile = "2.0" rustls-pki-types = "1.0" +ryu.workspace = true serde.workspace = true serde_json.workspace = true session.workspace = true diff --git a/src/servers/src/http/result/prometheus_resp.rs b/src/servers/src/http/result/prometheus_resp.rs index e79628fac0..bc1d084903 100644 --- a/src/servers/src/http/result/prometheus_resp.rs +++ b/src/servers/src/http/result/prometheus_resp.rs @@ -38,6 +38,7 @@ use datatypes::prelude::ConcreteDataType; use indexmap::IndexMap; use promql_parser::label::METRIC_NAME; use promql_parser::parser::value::ValueType; +use ryu::Buffer; use serde::{Deserialize, Serialize}; use serde_json::Value; use snafu::{OptionExt, ResultExt}; @@ -70,6 +71,19 @@ fn prometheus_native_histogram(histogram: &NativeHistogram) -> Result String { + if value.is_finite() { + Buffer::new().format_finite(value).to_string() + } else { + value.to_string() + } +} + #[derive(Debug, Default, Serialize, Deserialize, PartialEq)] pub struct PrometheusJsonResponse { pub status: String, @@ -307,6 +321,15 @@ impl PrometheusJsonResponse { // Tag order matters, e.g., after sorc and sort_desc, the output order must be kept. let mut buffer = IndexMap::, PromSeriesSamples>::new(); + // Query output is clustered by series (the range plan sorts by series + // key + timestamp), so consecutive rows usually belong to the same + // series. Remember the index of the previous row's entry in `buffer`, + // and reuse it directly when its tags are unchanged. This avoids + // building and hashing the label vector on every row; the worst case + // adds one `Vec` comparison per series transition before falling back + // to the map lookup. + let mut last_entry_index = None; + let schema = batches.schema(); for batch in batches.iter() { // prepare things... @@ -380,15 +403,32 @@ impl PrometheusJsonResponse { } } - let entry = buffer.entry(tags).or_default(); + let reuse = last_entry_index.filter(|index| { + buffer + .get_index(*index) + .is_some_and(|(key, _)| key == &tags) + }); + let samples = if let Some(index) = reuse { + buffer + .get_index_mut(index) + .map(|(_, samples)| samples) + .with_context(|| UnexpectedResultSnafu { + reason: "reused series entry must exist", + })? + } else { + let entry = buffer.entry(tags); + last_entry_index = Some(entry.index()); + entry.or_default() + }; if let Some((timestamp_millis, histogram)) = histogram { - entry + samples .histograms .push((timestamp_millis as f64 / 1000.0, histogram)); } else if let Some((timestamp_millis, value)) = value { - entry - .values - .push((timestamp_millis as f64 / 1000.0, value.to_string())); + samples.values.push(( + timestamp_millis as f64 / 1000.0, + format_prometheus_sample_value(value), + )); } } } @@ -564,6 +604,67 @@ mod tests { Arc::new(StructVector::try_new(histogram_type, histogram_array).unwrap()) } + #[test] + fn format_prometheus_sample_value_uses_ryu_for_finite_values() { + let values = [ + 1.5, + 0.1, + 1.0, + 0.0, + -0.0, + 100.0, + 1e-6, + 1e-7, + 1e21, + 1e30, + f64::MAX, + f64::MIN_POSITIVE, + ]; + + for value in values { + let output = format_prometheus_sample_value(value); + assert_eq!(output.parse::().unwrap().to_bits(), value.to_bits()); + } + + // Representative integral values use ryu's explicit .0 form. + assert_eq!(format_prometheus_sample_value(1.0), "1.0"); + assert_eq!(format_prometheus_sample_value(-0.0), "-0.0"); + assert_eq!(format_prometheus_sample_value(100.0), "100.0"); + assert_eq!(format_prometheus_sample_value(1e-6), "1e-6"); + assert_eq!(format_prometheus_sample_value(1e-7), "1e-7"); + assert_eq!(format_prometheus_sample_value(1e21), "1e21"); + + // These known shortest-roundtrip tie cases have different text but + // remain numerically equivalent to Rust's representation. + for value in [ + f64::from_bits(0x42374876e8000400), + f64::from_bits(0x3ff0000800000000), + f64::from_bits(0x430a8e5672bc7312), + ] { + let ryu_output = format_prometheus_sample_value(value); + let std_output = value.to_string(); + assert_ne!(ryu_output, std_output); + assert_eq!(ryu_output.parse::().unwrap(), value); + assert_eq!(std_output.parse::().unwrap(), value); + } + } + + #[test] + fn format_prometheus_sample_value_preserves_nonfinite_values() { + assert_eq!(format_prometheus_sample_value(f64::NAN), "NaN"); + assert_eq!(format_prometheus_sample_value(f64::INFINITY), "inf"); + assert_eq!(format_prometheus_sample_value(f64::NEG_INFINITY), "-inf"); + + // Parsing a NaN does not preserve its payload bits, so NaN is checked + // by its required semantic spelling rather than by to_bits(). + assert!( + format_prometheus_sample_value(f64::NAN) + .parse::() + .unwrap() + .is_nan() + ); + } + #[test] fn matrix_response_preserves_ordinary_nan_and_filters_stale_markers() { let schema = Arc::new(Schema::new(vec![ @@ -604,10 +705,190 @@ mod tests { assert_eq!(series.len(), 1); assert_eq!( series[0].values, - vec![(1.0, "1".to_string()), (2.0, "NaN".to_string())] + vec![(1.0, "1.0".to_string()), (2.0, "NaN".to_string())] ); } + #[test] + fn record_batches_to_data_formats_values_with_ryu() { + let schema = Arc::new(Schema::new(vec![ + ColumnSchema::new( + "timestamp", + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ), + ColumnSchema::new("value", ConcreteDataType::float64_datatype(), true), + ])); + let batch = RecordBatch::new( + schema.clone(), + vec![ + Arc::new(TimestampMillisecondVector::from_vec(vec![ + 1_000, 2_000, 3_000, 4_000, 5_000, 6_000, + ])) as _, + Arc::new(Float64Vector::from(vec![ + Some(0.0), + Some(-0.0), + Some(1.25), + Some(1e30), + Some(1e-7), + Some(f64::MAX), + ])) as _, + ], + ) + .unwrap(); + let batches = RecordBatches::try_new(schema, vec![batch]).unwrap(); + + let response = + PrometheusJsonResponse::record_batches_to_data(batches, None, ValueType::Matrix) + .unwrap(); + let PrometheusResponse::PromData(PromData { + result: PromQueryResult::Matrix(series), + .. + }) = response + else { + panic!("expected matrix response"); + }; + + assert_eq!(series.len(), 1); + let input_values = [0.0, -0.0, 1.25, 1e30, 1e-7, f64::MAX]; + // Keep this expected-value generation independent from the production + // formatter while still asserting the Arrow/batch-to-Prometheus path. + let expected = input_values + .into_iter() + .enumerate() + .map(|(index, value)| { + let expected_value = if value.is_finite() { + let mut buffer = Buffer::new(); + buffer.format_finite(value).to_string() + } else { + value.to_string() + }; + ((index + 1) as f64, expected_value) + }) + .collect::>(); + assert_eq!(series[0].values, expected); + } + + #[test] + fn record_batches_to_data_preserves_infinity_output() { + // NaN and infinities use Rust's `f64::to_string()` output. + let schema = Arc::new(Schema::new(vec![ + ColumnSchema::new( + "timestamp", + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ), + ColumnSchema::new("value", ConcreteDataType::float64_datatype(), true), + ])); + let batch = RecordBatch::new( + schema.clone(), + vec![ + Arc::new(TimestampMillisecondVector::from_vec(vec![ + 1_000, 2_000, 3_000, + ])) as _, + Arc::new(Float64Vector::from(vec![ + Some(f64::INFINITY), + Some(f64::NEG_INFINITY), + Some(f64::NAN), + ])) as _, + ], + ) + .unwrap(); + let batches = RecordBatches::try_new(schema, vec![batch]).unwrap(); + + let response = + PrometheusJsonResponse::record_batches_to_data(batches, None, ValueType::Matrix) + .unwrap(); + let PrometheusResponse::PromData(PromData { + result: PromQueryResult::Matrix(series), + .. + }) = response + else { + panic!("expected matrix response"); + }; + + assert_eq!(series.len(), 1); + assert_eq!( + series[0].values, + vec![ + (1.0, "inf".to_string()), + (2.0, "-inf".to_string()), + (3.0, "NaN".to_string()), + ] + ); + } + + #[test] + fn record_batches_to_data_reuses_entries_for_clustered_series() { + // Rows are clustered by series (a, b, a, a, b, c): consecutive rows of + // the same series exercise the entry-reuse fast path, while series + // transitions fall back to the map lookup. The result must keep the + // first-occurrence order and accumulate values per series as before. + let schema = Arc::new(Schema::new(vec![ + ColumnSchema::new( + "timestamp", + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ), + ColumnSchema::new("host", ConcreteDataType::string_datatype(), false), + ColumnSchema::new("value", ConcreteDataType::float64_datatype(), true), + ])); + let batch = RecordBatch::new( + schema.clone(), + vec![ + Arc::new(TimestampMillisecondVector::from_vec(vec![ + 1_000, 2_000, 3_000, 4_000, 5_000, 6_000, + ])) as _, + Arc::new(StringVector::from(vec![ + Some("a"), + Some("b"), + Some("a"), + Some("a"), + Some("b"), + Some("c"), + ])) as _, + Arc::new(Float64Vector::from(vec![ + Some(1.0), + Some(2.0), + Some(3.0), + Some(4.0), + Some(5.0), + Some(6.0), + ])) as _, + ], + ) + .unwrap(); + let batches = RecordBatches::try_new(schema, vec![batch]).unwrap(); + + let response = PrometheusJsonResponse::record_batches_to_data( + batches, + Some("metric".to_string()), + ValueType::Vector, + ) + .unwrap(); + let PrometheusResponse::PromData(PromData { + result: PromQueryResult::Vector(series), + .. + }) = response + else { + panic!("expected vector response"); + }; + + assert_eq!(series.len(), 3); + // Output order is first-occurrence order: a, b, c. + assert_eq!( + series + .iter() + .map(|series| series.metric["host"].as_str()) + .collect::>(), + vec!["a", "b", "c"] + ); + // Vector results keep the last sample of each series. + assert_eq!(series[0].value, Some((4.0, "4.0".to_string()))); + assert_eq!(series[1].value, Some((5.0, "5.0".to_string()))); + assert_eq!(series[2].value, Some((6.0, "6.0".to_string()))); + } + #[test] fn record_batches_to_data_preserves_mixed_float_and_histogram_rows() { let schema = Arc::new(Schema::new(vec![ @@ -713,7 +994,7 @@ mod tests { assert_eq!(series[0].metric["__name__"], "label_replace_repro"); assert_eq!(series[0].metric["host"], "server-01"); assert_eq!(series[0].metric["host_copy"], "server-01"); - assert_eq!(series[0].value, Some((1.0, "1".to_string()))); + assert_eq!(series[0].value, Some((1.0, "1.0".to_string()))); } #[test] diff --git a/tests-integration/tests/grpc.rs b/tests-integration/tests/grpc.rs index 9d0e13dddc..737b4fb6c1 100644 --- a/tests-integration/tests/grpc.rs +++ b/tests-integration/tests/grpc.rs @@ -1071,7 +1071,7 @@ pub async fn test_prom_gateway_query(store_type: StorageType) { ] .into_iter() .collect(), - value: Some((5.0, "1".to_string())), + value: Some((5.0, "1.0".to_string())), ..Default::default() }, PromSeriesVector { @@ -1081,7 +1081,7 @@ pub async fn test_prom_gateway_query(store_type: StorageType) { ] .into_iter() .collect(), - value: Some((5.0, "2".to_string())), + value: Some((5.0, "2.0".to_string())), ..Default::default() }, ] @@ -1133,7 +1133,7 @@ pub async fn test_prom_gateway_query(store_type: StorageType) { ] .into_iter() .collect(), - values: vec![(5.0, "1".to_string()), (10.0, "1".to_string())], + values: vec![(5.0, "1.0".to_string()), (10.0, "1.0".to_string())], ..Default::default() }, PromSeriesMatrix { @@ -1143,7 +1143,7 @@ pub async fn test_prom_gateway_query(store_type: StorageType) { ] .into_iter() .collect(), - values: vec![(5.0, "2".to_string()), (10.0, "2".to_string())], + values: vec![(5.0, "2.0".to_string()), (10.0, "2.0".to_string())], ..Default::default() }, ] diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 2a66c4353e..7a78783337 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -935,7 +935,7 @@ pub async fn test_prom_http_api(store_type: StorageType) { assert_eq!( body.data, serde_json::from_value::( - json!({"resultType":"scalar","result":[1.0,"2"]}) + json!({"resultType":"scalar","result":[1.0,"2.0"]}) ) .unwrap() ); @@ -2967,6 +2967,8 @@ pub async fn test_prometheus_remote_write_v2_native_histogram(store_type: Storag .await; assert_eq!(res.status(), StatusCode::OK); let body = res.json::().await; + // Finite vector sample values use the HTTP response's ryu contract, so + // integral f64 values retain their explicit `.0`; timestamps remain JSON numbers. assert_eq!( serde_json::to_string_pretty(&body).unwrap(), r#"{ @@ -2982,7 +2984,7 @@ pub async fn test_prometheus_remote_write_v2_native_histogram(store_type: Storag }, "value": [ 4.0, - "6" + "6.0" ] } ] diff --git a/tests/perf/fixture-format.md b/tests/perf/fixture-format.md index ca60ae54d7..361f0ec79f 100644 --- a/tests/perf/fixture-format.md +++ b/tests/perf/fixture-format.md @@ -58,6 +58,20 @@ warmup = 0 iterations = 1 ``` +### Query kinds + +`kind = "prom_http"` runs a Prometheus range query by POSTing form fields to +`/v1/prometheus/api/v1/query_range`. `query`, `start`, `end`, and `step` are +sent as form fields; `start` and `end` define the range, and `step` defines its +evaluation interval. `start`, `end`, and `step` are optional in the case model +and default to empty form values when omitted. The table database is sent as the +`db` URL query parameter, rather than as a form field. + +The measurement starts before the HTTP request is sent and ends after the full +response body is read and parsed as JSON. It therefore includes request send, +server-side Prometheus JSON response building, and response-body read and JSON +parsing; it is not server-side-only latency. + `series_layout = "round_robin"` advances the timestamp once per generated row and cycles series labels across rows. `series_layout = "timestamp_major"` writes all series for one timestamp before advancing to the next timestamp; use it for diff --git a/tests/perf/query_cases/prom_json_response/case.toml b/tests/perf/query_cases/prom_json_response/case.toml new file mode 100644 index 0000000000..17472cd5f3 --- /dev/null +++ b/tests/perf/query_cases/prom_json_response/case.toml @@ -0,0 +1,89 @@ +# Prometheus JSON response building benchmark (issue #8805). +# +# Exercises the serial per-request +# `PrometheusJsonResponse::record_batches_to_data` +# (src/servers/src/http/result/prometheus_resp.rs) path, ryu float formatting, +# and its per-series IndexMap entry reuse. +# +# 256 series x 481 evaluation timestamps (2h at 15s) = ~123k output samples +# per request. The primary `prom_http` query builds the Prometheus JSON +# response. The TQL ANALYZE control runs the same workload without building the +# Prometheus JSON response, to measure query-engine/scan variance. + +[case] +name = "prom_json_response" +description = "Prometheus HTTP API JSON response building for large range queries" + +[scenario] +kind = "direct_readable_sst" +seed = 8805 + +[[scenario.tables]] +database = "public" +name = "prom_json_response" +engine = "mito" +append_mode = true +sst_format = "flat" +primary_key = ["host", "instance"] +time_index = "ts" + +[[scenario.tables.columns]] +name = "host" +type = "STRING" +semantic = "tag" +distribution = { kind = "cardinality", values = 16, prefix = "host" } + +[[scenario.tables.columns]] +name = "instance" +type = "STRING" +semantic = "tag" +distribution = { kind = "cardinality", values = 256, prefix = "instance" } + +[[scenario.tables.columns]] +name = "value" +type = "DOUBLE" +semantic = "field" +distribution = { kind = "deterministic_wave", min = 0.0, max = 1000.0 } + +[[scenario.tables.columns]] +name = "ts" +type = "TIMESTAMP(9)" +semantic = "timestamp" + +[scenario.layout] +regions = 1 +sst_count = 16 +rows_per_sst = 7680 +row_group_size = 1920 +series_count = 256 +start_unix_nanos = 1704067200000000000 +step_nanos = 15000000000 +time_range_layout = "non_overlapping_per_sst" +series_layout = "timestamp_major" + +# Primary case: Prometheus HTTP range query over the full 2h window. +# 256 series x 481 points -> ~123k samples built and serialized per request. +[[scenario.queries]] +name = "prom_range_2h" +kind = "prom_http" +query = "prom_json_response{host=~\"host.*\"}" +start = "1704067200" +end = "1704074400" +step = "15s" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +# Control: same workload through the SQL/TQL frontend path, which does not +# build the Prometheus JSON response. Isolates query-engine/scan variance. +[[scenario.queries]] +name = "tql_range_2h_control" +kind = "tql" +query = "TQL ANALYZE VERBOSE (1704067200, 1704074400, '15s') prom_json_response{host=~'host.*'}" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10