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

perf(servers)!: speed up Prometheus JSON response building with ryu and per-series entry reuse (#8815)

Browse Source 
* fix(perf): align direct-SST CREATE TABLE with baked index metadata

The offline fixture generator (query_perf_fixture::direct_sst::
build_region_metadata) bakes greptime:inverted_index /
greptime:skipping_index field metadata into the region manifest for
tag/field columns, but create_table_sql emitted a bare CREATE TABLE
without those declarations. MergeScan's remote-schema validation then
failed on any tag/field projection (HTTP 500 'advertised remote stream
schema field mismatch'), breaking direct_readable_sst perf cases.

CREATE TABLE now declares the matching SKIPPING INDEX WITH
(granularity='1') / INVERTED INDEX column options. A round-trip test
proves the emitted SQL is parser-valid and yields the exact catalog
metadata.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* perf(servers): speed up Prometheus JSON response building with ryu and per-series entry reuse

PrometheusJsonResponse::record_batches_to_data spends ~47% of its CPU
in f64::to_string() per sample and ~32% in IndexMap::entry() per row
(60s profile of concurrent query_range workloads, ~800k series).

- Replace f64::to_string() with ryu::Buffer::format_finite for finite
  values (shortest round-trip, 3-5x faster); NaN/+Inf/-Inf keep the
  previous std formatting so wire output is unchanged.
- Remember the previous row's label vector and entry index; query output
  is clustered by series, so consecutive rows reuse the same IndexMap
  entry via get_index_mut instead of rebuilding and hashing the label
  vector (worst case adds one Vec comparison per series transition).

Also adds a query-regression case (prom_json_response) that measures the
real Prometheus HTTP range API path (/v1/prometheus/api/v1/query_range),
which is the only frontend path that builds the Prometheus JSON response
(TQL ANALYZE formats the SQL JSON shape instead), plus a prom_http query
kind in the regression runner.

Perf (aligned base d90cca4b75, 256 series x 481 points):
- prom_range_2h (JSON response path): 29.31ms -> 21.46ms (-26.8%)
- tql_range_2h_control (non-JSON path): +1.87% (noise)

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(servers): keep Prometheus wire format for integral floats with ryu

ryu::Buffer::format_finite prints integral values as "1.0", but the
Prometheus JSON wire format (matching std f64::to_string) expects "1".
Strip the trailing ".0" that ryu only emits for integral values; extreme
values keep ryu scientific notation, and NaN/Inf keep std output. Adds
wire-format tests covering 1.0, 0.0, -0.0, 1.5, 0.1, 1e21, 1e30, 1e-7,
f64::MAX, NaN, ±Inf.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(servers): address Prometheus response review feedback

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* perf(servers)!: use ryu for Prometheus sample values

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(cmd): skip Prometheus execution time extraction

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

---------

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-08-25 05:16:50 +00:00
committed by GitHub
parent 0cc83c4570
commit 3493d2d0fb
12 changed files with 644 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -13470,6 +13470,7 @@ dependencies = [
"rustls",
"rustls-pemfile",
"rustls-pki-types",
"ryu",
"serde",
"serde_json",
"servers",
Cargo.toml
+1
View File
@@ -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"] }
src/cmd/src/bin/query_regression_runner/direct.rs
+104 -2
View File
@@ -166,7 +166,28 @@ fn create_table_sql(table: &Table) -> Result<String> {
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::<Vec<_>>()
.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;"
);
}
src/cmd/src/bin/query_regression_runner/measure.rs
+59 -5
View File
@@ -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([
src/cmd/src/bin/query_regression_runner/model.rs
+13
View File
@@ -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<String>,
}
const fn default_show_create_engine() -> bool {
@@ -203,6 +209,13 @@ pub(super) struct Query {
#[serde(default)]
pub(super) kind: Option<String>,
pub(super) query: String,
/// Prometheus HTTP range query parameters (`kind = "prom_http"` only).
#[serde(default)]
pub(super) start: Option<String>,
#[serde(default)]
pub(super) end: Option<String>,
#[serde(default)]
pub(super) step: Option<String>,
#[serde(default)]
pub(super) warmup: usize,
#[serde(default = "one")]
src/cmd/src/bin/query_regression_runner/sql.rs
+66 -6
View File
@@ -117,10 +117,62 @@ pub(super) fn value_f64(value: Option<&Value>) -> Option<f64> {
}
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<_>>(),
vec![("db".into(), "catalog-schema name".into())]
);
}
#[test]
fn top_level_errors_do_not_inspect_rows() {
assert!(response_has_error(&json!({"error_code": 7})));
src/servers/Cargo.toml
+1
View File
@@ -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
src/servers/src/http/result/prometheus_resp.rs
+288 -7
View File
@@ -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<PromNative
})
}
/// Formats a sample value for the Prometheus HTTP API.
///
/// Finite sample strings use ryu's shortest-roundtrip representation and may
/// differ textually from previous Rust/Prometheus wire formatting; values parse
/// identically. Non-finite values retain Rust's `f64::to_string()` output.
fn format_prometheus_sample_value(value: f64) -> 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::<Vec<(&str, &str)>, 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::<f64>().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::<f64>().unwrap(), value);
assert_eq!(std_output.parse::<f64>().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::<f64>()
.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::<Vec<_>>();
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<_>>(),
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]
tests-integration/tests/grpc.rs
+4 -4
View File
@@ -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()
},
]
tests-integration/tests/http.rs
+4 -2
View File
@@ -935,7 +935,7 @@ pub async fn test_prom_http_api(store_type: StorageType) {
assert_eq!(
body.data,
serde_json::from_value::<PrometheusResponse>(
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::<PrometheusJsonResponse>().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"
]
}
]
tests/perf/fixture-format.md
+14
View File
@@ -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
tests/perf/query_cases/prom_json_response/case.toml
+89
View File
@@ -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