diff --git a/src/cmd/src/bin/query_perf_fixture/case.rs b/src/cmd/src/bin/query_perf_fixture/case.rs index a76c49d43d..5ed2466999 100644 --- a/src/cmd/src/bin/query_perf_fixture/case.rs +++ b/src/cmd/src/bin/query_perf_fixture/case.rs @@ -80,6 +80,8 @@ pub(super) struct OtlpTraceLoadThresholds { #[derive(Debug, Deserialize, Serialize)] pub(super) struct PromRemoteWritePlan { + #[serde(default)] + pub(super) input_protocol: InputProtocol, #[serde(default = "default_database")] pub(super) database: String, #[serde(alias = "metric_name")] @@ -114,6 +116,14 @@ pub(super) struct PromRemoteWritePlan { pub(super) read_bench: Option, } +#[derive(Debug, Default, Deserialize, Serialize, Eq, PartialEq)] +#[serde(rename_all = "snake_case")] +pub(super) enum InputProtocol { + #[default] + RemoteWrite, + OtlpMetrics, +} + pub(super) fn default_database() -> String { "public".to_string() } diff --git a/src/cmd/src/bin/query_perf_fixture/mod.rs b/src/cmd/src/bin/query_perf_fixture/mod.rs index 17f639bb54..9694d6d017 100644 --- a/src/cmd/src/bin/query_perf_fixture/mod.rs +++ b/src/cmd/src/bin/query_perf_fixture/mod.rs @@ -15,6 +15,7 @@ mod case; mod direct_sst; mod inspect_footer; +mod otlp_metrics; mod prom_remote_write; mod util; @@ -25,6 +26,7 @@ use case::*; use clap::{Args as ClapArgs, Parser, Subcommand}; use direct_sst::run_direct_sst; use inspect_footer::{InspectFooterArgs, run_inspect_footer}; +use otlp_metrics::{OtlpMetricsArgs, run_otlp_metrics}; use prom_remote_write::{PromRemoteWriteArgs, run_prom_remote_write}; use serde_json::json; @@ -43,6 +45,7 @@ struct Args { enum Command { DirectSst(DirectArgs), PromRemoteWrite(PromRemoteWriteArgs), + OtlpMetrics(OtlpMetricsArgs), InspectFooter(InspectFooterArgs), Plan(PlanArgs), } @@ -95,6 +98,9 @@ pub async fn run() { Some(Command::PromRemoteWrite(rw)) => run_prom_remote_write(rw) .await .expect("prom remote write failed"), + Some(Command::OtlpMetrics(otlp)) => run_otlp_metrics(otlp) + .await + .expect("OTLP metrics export failed"), Some(Command::InspectFooter(inspect)) => run_inspect_footer(inspect) .await .expect("inspect footer failed"), diff --git a/src/cmd/src/bin/query_perf_fixture/otlp_metrics.rs b/src/cmd/src/bin/query_perf_fixture/otlp_metrics.rs new file mode 100644 index 0000000000..c49fbf8811 --- /dev/null +++ b/src/cmd/src/bin/query_perf_fixture/otlp_metrics.rs @@ -0,0 +1,294 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use clap::Args as ClapArgs; +use prost::Message; +use serde_json::json; + +use crate::query_perf_fixture::case::ValuePattern; +use crate::query_perf_fixture::prom_remote_write::deterministic_value; + +#[derive(Debug, ClapArgs)] +pub(super) struct OtlpMetricsArgs { + #[arg(long, default_value = "http://127.0.0.1:4000/v1/otlp/v1/metrics")] + endpoint: String, + #[arg(long, default_value = "public")] + database: String, + #[arg(long)] + metric: String, + #[arg(long, default_value_t = 8)] + series_count: u64, + #[arg(long, default_value_t = 30)] + samples_per_series: u64, + #[arg(long)] + start_unix_nanos: u64, + #[arg(long)] + step_nanos: u64, + #[arg(long, alias = "batch-size", default_value_t = 8)] + chunk_series_count: u64, + #[arg(long, default_value_t = 60)] + timeout_seconds: u64, + #[arg(long, default_value_t = ValuePattern::Linear)] + value_pattern: ValuePattern, + #[arg(long, default_value_t = 0.0)] + value_base: f64, + #[arg(long, default_value_t = 0.125)] + value_step: f64, + #[arg(long, default_value_t = 97)] + value_cardinality: u64, + #[arg(long, default_value_t = 0)] + value_seed: u64, + #[arg(long, default_value_t = 8)] + value_run_length: u64, + #[arg(long, default_value_t = 100)] + value_stall_every: u64, + #[arg(long, default_value_t = 16)] + value_stall_length: u64, + #[arg(long, default_value_t = 5)] + value_mixed_every: u64, +} + +pub(super) async fn run_otlp_metrics( + args: OtlpMetricsArgs, +) -> Result<(), Box> { + let started = std::time::Instant::now(); + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(args.timeout_seconds)) + .build()?; + let mut batches = 0_u64; + let mut rows = 0_u64; + let mut http_statuses = Vec::new(); + let mut rejected_data_points = 0_i64; + for first in (0..args.series_count).step_by(args.chunk_series_count.max(1) as usize) { + let last = (first + args.chunk_series_count.max(1)).min(args.series_count); + let request = ExportMetricsServiceRequest { + resource_metrics: vec![ResourceMetrics { + scope_metrics: vec![ScopeMetrics { + metrics: vec![Metric { + name: args.metric.clone(), + data: Some(metric::Data::Gauge(Gauge { + data_points: (first..last) + .flat_map(|series_idx| { + (0..args.samples_per_series).map(move |sample_idx| { + NumberDataPoint { + attributes: vec![ + string_attribute( + "host", + format!("host{:04}", series_idx % 1024), + ), + string_attribute( + "instance", + format!("instance{:06}", series_idx), + ), + ], + time_unix_nano: args.start_unix_nanos + + sample_idx * args.step_nanos, + value: Some(number_data_point::Value::AsDouble( + deterministic_value( + args.value_pattern, + args.value_base, + args.value_step, + args.value_cardinality, + args.value_seed, + args.value_run_length, + args.value_stall_every, + args.value_stall_length, + args.value_mixed_every, + series_idx, + sample_idx, + args.samples_per_series, + ), + )), + } + }) + }) + .collect(), + })), + }], + }], + }], + }; + let batch_rows = (last - first) * args.samples_per_series; + let response = client + .post(&args.endpoint) + .header("content-type", "application/x-protobuf") + .header("x-greptime-db-name", &args.database) + .body(request.encode_to_vec()) + .send() + .await?; + let status = response.status(); + let bytes = response.bytes().await?; + http_statuses.push(status.as_u16()); + if !status.is_success() { + return Err(format!( + "OTLP metrics export failed with status {status}: {}", + String::from_utf8_lossy(&bytes) + ) + .into()); + } + let response = ExportMetricsServiceResponse::decode(bytes.as_ref())?; + let rejected = response + .partial_success + .map(|partial| partial.rejected_data_points) + .unwrap_or_default(); + if rejected != 0 { + return Err(format!("OTLP metrics export rejected {rejected} data points").into()); + } + rejected_data_points += rejected; + rows += batch_rows; + batches += 1; + } + println!( + "{}", + json!({"status":"ok","endpoint":args.endpoint,"database":args.database,"metric":args.metric,"input_protocol":"otlp_metrics","timestamp_precision":"ns","series_count":args.series_count,"samples_per_series":args.samples_per_series,"rows":rows,"samples_written":rows,"batches":batches,"elapsed_seconds":started.elapsed().as_secs_f64(),"http_statuses":http_statuses,"rejected_data_points":rejected_data_points}) + ); + Ok(()) +} + +fn string_attribute(key: &str, value: String) -> KeyValue { + KeyValue { + key: key.to_string(), + value: Some(AnyValue { + value: Some(any_value::Value::StringValue(value)), + }), + } +} + +// The metrics endpoint uses the otel-arrow generated metric messages. Keep this +// small wire encoder local to the fixture binary so production protocol code is +// untouched. +#[derive(Clone, PartialEq, Message)] +struct ExportMetricsServiceRequest { + #[prost(message, repeated, tag = "1")] + resource_metrics: Vec, +} + +#[derive(Clone, PartialEq, Message)] +struct ResourceMetrics { + #[prost(message, repeated, tag = "2")] + scope_metrics: Vec, +} + +#[derive(Clone, PartialEq, Message)] +struct ScopeMetrics { + #[prost(message, repeated, tag = "2")] + metrics: Vec, +} + +#[derive(Clone, PartialEq, Message)] +struct Metric { + #[prost(string, tag = "1")] + name: String, + #[prost(oneof = "metric::Data", tags = "5")] + data: Option, +} + +mod metric { + #[derive(Clone, PartialEq, prost::Oneof)] + pub(super) enum Data { + #[prost(message, tag = "5")] + Gauge(super::Gauge), + } +} + +#[derive(Clone, PartialEq, Message)] +struct Gauge { + #[prost(message, repeated, tag = "1")] + data_points: Vec, +} + +#[derive(Clone, PartialEq, Message)] +struct NumberDataPoint { + #[prost(message, repeated, tag = "7")] + attributes: Vec, + #[prost(fixed64, tag = "3")] + time_unix_nano: u64, + #[prost(oneof = "number_data_point::Value", tags = "4")] + value: Option, +} + +mod number_data_point { + #[derive(Clone, Copy, PartialEq, prost::Oneof)] + pub(super) enum Value { + #[prost(double, tag = "4")] + AsDouble(f64), + } +} + +#[derive(Clone, PartialEq, Message)] +struct KeyValue { + #[prost(string, tag = "1")] + key: String, + #[prost(message, optional, tag = "2")] + value: Option, +} + +#[derive(Clone, PartialEq, Message)] +struct AnyValue { + #[prost(oneof = "any_value::Value", tags = "1")] + value: Option, +} + +mod any_value { + #[derive(Clone, PartialEq, prost::Oneof)] + pub(super) enum Value { + #[prost(string, tag = "1")] + StringValue(String), + } +} + +#[derive(Clone, PartialEq, Message)] +struct ExportMetricsServiceResponse { + #[prost(message, optional, tag = "1")] + partial_success: Option, +} + +#[derive(Clone, PartialEq, Message)] +struct ExportMetricsPartialSuccess { + #[prost(int64, tag = "1")] + rejected_data_points: i64, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn encodes_gauge_timestamp_with_nanosecond_remainder() { + let point = NumberDataPoint { + attributes: vec![string_attribute("host", "host0000".to_string())], + time_unix_nano: 1_704_067_200_000_000_123, + value: Some(number_data_point::Value::AsDouble(1.0)), + }; + let decoded = NumberDataPoint::decode(point.encode_to_vec().as_slice()).unwrap(); + assert_eq!(decoded.time_unix_nano % 1_000_000_000, 123); + } + + #[test] + fn decodes_partial_success_rejections() { + let response = ExportMetricsServiceResponse { + partial_success: Some(ExportMetricsPartialSuccess { + rejected_data_points: 2, + }), + }; + assert_eq!( + ExportMetricsServiceResponse::decode(response.encode_to_vec().as_slice()) + .unwrap() + .partial_success + .unwrap() + .rejected_data_points, + 2 + ); + } +} diff --git a/src/cmd/src/bin/query_perf_fixture/prom_remote_write.rs b/src/cmd/src/bin/query_perf_fixture/prom_remote_write.rs index 1f36b0c2d2..7568b182dc 100644 --- a/src/cmd/src/bin/query_perf_fixture/prom_remote_write.rs +++ b/src/cmd/src/bin/query_perf_fixture/prom_remote_write.rs @@ -147,58 +147,74 @@ fn prom_label(name: &str, value: &str) -> api::prom_store::remote::Label { } } fn deterministic_prom_value(args: &PromRemoteWriteArgs, series_idx: u64, sample_idx: u64) -> f64 { - let ordinal = series_idx - * args - .value_total_samples_per_series - .unwrap_or(args.samples_per_series) - + args.value_sample_offset - + sample_idx; - let local = args.value_sample_offset + sample_idx; - match args.value_pattern { - ValuePattern::Linear => { - args.value_base + (series_idx % 97) as f64 + local as f64 * args.value_step - } - ValuePattern::Constant => args.value_base, - ValuePattern::Modulo => { - args.value_base + (ordinal % args.value_cardinality.max(1)) as f64 * args.value_step - } - ValuePattern::Unique => args.value_base + ordinal as f64 * args.value_step, + deterministic_value( + args.value_pattern, + args.value_base, + args.value_step, + args.value_cardinality, + args.value_seed, + args.value_run_length, + args.value_stall_every, + args.value_stall_length, + args.value_mixed_every, + series_idx, + args.value_sample_offset + sample_idx, + args.value_total_samples_per_series + .unwrap_or(args.samples_per_series), + ) +} + +pub(super) fn deterministic_value( + pattern: ValuePattern, + base: f64, + step: f64, + cardinality: u64, + seed: u64, + run_length: u64, + stall_every: u64, + stall_length: u64, + mixed_every: u64, + series_idx: u64, + sample_idx: u64, + total_samples_per_series: u64, +) -> f64 { + let ordinal = series_idx * total_samples_per_series + sample_idx; + match pattern { + ValuePattern::Linear => base + (series_idx % 97) as f64 + sample_idx as f64 * step, + ValuePattern::Constant => base, + ValuePattern::Modulo => base + (ordinal % cardinality.max(1)) as f64 * step, + ValuePattern::Unique => base + ordinal as f64 * step, ValuePattern::SeededRandom => { - args.value_base - + (splitmix64(ordinal ^ args.value_seed) % args.value_cardinality.max(1)) as f64 - * args.value_step + base + (splitmix64(ordinal ^ seed) % cardinality.max(1)) as f64 * step } ValuePattern::RunLength => { - args.value_base - + ((ordinal / args.value_run_length.max(1)) % args.value_cardinality.max(1)) as f64 - * args.value_step + base + ((ordinal / run_length.max(1)) % cardinality.max(1)) as f64 * step } ValuePattern::QuantizedSignal => { - args.value_base - + (((series_idx % args.value_cardinality.max(1)) - + (local / args.value_run_length.max(1))) - % args.value_cardinality.max(1)) as f64 - * args.value_step + base + (((series_idx % cardinality.max(1)) + (sample_idx / run_length.max(1))) + % cardinality.max(1)) as f64 + * step } ValuePattern::SignalWithSporadicStalls => { - let every = args.value_stall_every.max(1); - let phase = local % every; - let effective = if phase < args.value_stall_length.min(every) { - local - phase + let every = stall_every.max(1); + let phase = sample_idx % every; + let effective = if phase < stall_length.min(every) { + sample_idx - phase } else { - local + sample_idx }; - args.value_base + (series_idx % 97) as f64 + effective as f64 * args.value_step + base + (series_idx % 97) as f64 + effective as f64 * step } ValuePattern::MixedSignalRepeated => { - if local.is_multiple_of(args.value_mixed_every.max(1)) { - args.value_base + if sample_idx.is_multiple_of(mixed_every.max(1)) { + base } else { - args.value_base + (series_idx % 97) as f64 + local as f64 * args.value_step + base + (series_idx % 97) as f64 + sample_idx as f64 * step } } } } + fn splitmix64(mut value: u64) -> u64 { value = value.wrapping_add(0x9e3779b97f4a7c15); let mut mixed = value; diff --git a/src/cmd/src/bin/query_regression_runner/direct.rs b/src/cmd/src/bin/query_regression_runner/direct.rs index b003b6ee5d..431b86335b 100644 --- a/src/cmd/src/bin/query_regression_runner/direct.rs +++ b/src/cmd/src/bin/query_regression_runner/direct.rs @@ -489,6 +489,7 @@ mod tests { append_mode: Some(true), sst_format: Some("flat".to_string()), validate_show_create_engine: true, + validate_timestamp_nanos: None, }; assert_eq!( create_table_sql(&table).unwrap(), @@ -536,6 +537,7 @@ mod tests { append_mode: Some(true), sst_format: Some("flat".to_string()), validate_show_create_engine: true, + validate_timestamp_nanos: None, }; assert_eq!( create_table_sql(&table).unwrap(), @@ -568,6 +570,7 @@ mod tests { append_mode: None, sst_format: None, validate_show_create_engine: true, + validate_timestamp_nanos: None, }; assert_eq!( create_table_sql(&table).unwrap(), diff --git a/src/cmd/src/bin/query_regression_runner/measure.rs b/src/cmd/src/bin/query_regression_runner/measure.rs index fd80813f40..9f2bdc6975 100644 --- a/src/cmd/src/bin/query_regression_runner/measure.rs +++ b/src/cmd/src/bin/query_regression_runner/measure.rs @@ -49,14 +49,23 @@ pub(super) async fn run_measure(args: MeasureArgs) -> Result<()> { let client = Client::builder() .timeout(Duration::from_secs_f64(args.http_timeout)) .build()?; - let base = run_target(args.base_http_port, &tables, &configured_queries, &client).await; - let candidate = run_target( + let mut base = run_target( + args.base_http_port, + &tables, + &configured_queries, + &client, + false, + ) + .await; + let mut candidate = run_target( args.candidate_http_port, &tables, &configured_queries, &client, + true, ) .await; + compare_measured_results(&configured_queries, &mut base, &mut candidate); let thresholds = enforce_thresholds(&configured_queries, &base, &candidate)?; let status = if base.status == "failed" || candidate.status == "failed" @@ -74,6 +83,7 @@ pub(super) async fn run_measure(args: MeasureArgs) -> Result<()> { "scenario": scenario_value, "queries": configured_queries, "query_mode": "endpoint", + "input_protocol": scenario_value.pointer("/remote_write/input_protocol").cloned().unwrap_or(Value::String("direct_sst".to_string())), "http_timeout": args.http_timeout, "targets": [target_report("base", args.base_http_port, base), target_report("candidate", args.candidate_http_port, candidate)], "thresholds": thresholds, @@ -131,6 +141,7 @@ async fn run_target( tables: &[Table], configured_queries: &[Query], client: &Client, + validate_candidate_plan: bool, ) -> QueryResult { let mut queries = configured_queries.to_vec(); if queries.is_empty() { @@ -144,6 +155,11 @@ async fn run_target( warmup: 0, iterations: 1, thresholds: Map::new(), + expected_cardinality: None, + compare_results: false, + measure: true, + plan_contains: None, + plan_absent: None, }); } @@ -170,18 +186,30 @@ async fn run_target( } validation.push(sample); } - 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, - "error": first.get("error"), - "response": first.get("response"), - })); - } - validation.push(first); - let mut measurements = Vec::with_capacity(queries.len()); for query in &queries { + if !query.measure { + let sample = post_query(client, port, query, db).await; + if !sample["ok"].as_bool().unwrap_or(false) { + validation_errors.push(json!({"sql": query.query, "error": sample.get("error"), "response": sample.get("response")})); + } + if validate_candidate_plan + && let Some(plan_contains) = &query.plan_contains + && !response_text(sample.get("response").unwrap_or(&Value::Null)) + .contains(plan_contains) + { + validation_errors.push(json!({"sql": query.query, "error": format!("plan evidence does not contain {plan_contains}"), "response": sample.get("response")})); + } + if validate_candidate_plan + && let Some(plan_absent) = &query.plan_absent + && response_text(sample.get("response").unwrap_or(&Value::Null)) + .contains(plan_absent) + { + validation_errors.push(json!({"sql": query.query, "error": format!("plan evidence unexpectedly contains {plan_absent}"), "response": sample.get("response")})); + } + validation.push(sample); + continue; + } for _ in 0..query.warmup { let warmup = post_query(client, port, query, db).await; if !warmup["ok"].as_bool().unwrap_or(false) { @@ -239,6 +267,92 @@ async fn run_target( } } +fn compare_measured_results( + queries: &[Query], + base: &mut QueryResult, + candidate: &mut QueryResult, +) { + for query in queries { + if !query.measure || (!query.compare_results && query.expected_cardinality.is_none()) { + continue; + } + let Some(base_measurement) = base + .measurements + .iter() + .find(|measurement| measurement.name == query.name) + else { + continue; + }; + let Some(candidate_measurement) = candidate + .measurements + .iter() + .find(|measurement| measurement.name == query.name) + else { + continue; + }; + for (iteration, (base_sample, candidate_sample)) in base_measurement + .samples + .iter() + .zip(&candidate_measurement.samples) + .enumerate() + { + let base_rows = crate::query_regression_runner::sql::extract_rows( + base_sample.get("response").unwrap_or(&Value::Null), + ); + let candidate_rows = crate::query_regression_runner::sql::extract_rows( + candidate_sample.get("response").unwrap_or(&Value::Null), + ); + if let Some(expected) = query.expected_cardinality + && (base_rows.len() as u64 != expected || candidate_rows.len() as u64 != expected) + { + candidate.validation_errors.push(json!({"query": query.name, "iteration": iteration, "error": format!("expected cardinality {expected}, got base={} candidate={}", base_rows.len(), candidate_rows.len())})); + } + if query.compare_results + && normalized_rows(base_rows) != normalized_rows(candidate_rows) + { + candidate.validation_errors.push(json!({"query": query.name, "iteration": iteration, "error": "base/candidate response rows differ"})); + } + } + } + if !candidate.validation_errors.is_empty() { + candidate.status = "failed".to_string(); + } +} + +fn normalized_rows(rows: Vec) -> Vec { + let mut rows = rows + .into_iter() + .map(|mut row| { + strip_timing_metadata(&mut row); + serde_json::to_string(&row).unwrap_or_default() + }) + .collect::>(); + rows.sort_unstable(); + rows +} + +fn strip_timing_metadata(value: &mut Value) { + match value { + Value::Object(values) => { + values.retain(|key, _| { + !matches!( + key.as_str(), + "execution_time" | "execution_time_ms" | "elapsed" | "elapsed_ms" | "cost" + ) + }); + for value in values.values_mut() { + strip_timing_metadata(value); + } + } + Value::Array(values) => { + for value in values { + strip_timing_metadata(value); + } + } + _ => {} + } +} + fn validate_show_create(result: &Value, table: &Table) -> Vec<&'static str> { let text = result .get("response") @@ -258,6 +372,15 @@ fn validate_show_create(result: &Value, table: &Table) -> Vec<&'static str> { if table.sst_format.is_some() && !text.contains("sst_format") { errors.push("SHOW CREATE output does not mention sst_format"); } + if table.validate_timestamp_nanos.is_some() + && (!text.contains("greptime_timestamp") + || !text.contains("timestamp(9)") + || !text.contains("primary key") + || !text.contains("host") + || !text.contains("instance")) + { + errors.push("SHOW CREATE output does not contain the expected native OTLP Mito schema"); + } errors } @@ -441,6 +564,11 @@ mod tests { step: None, warmup: 0, iterations: 1, + expected_cardinality: None, + compare_results: false, + measure: true, + plan_contains: None, + plan_absent: None, thresholds: Map::from_iter([ ("max_candidate_latency_regression_pct".to_string(), json!(0)), ("other".to_string(), json!(1)), diff --git a/src/cmd/src/bin/query_regression_runner/model.rs b/src/cmd/src/bin/query_regression_runner/model.rs index 6831a5256a..38f6599fed 100644 --- a/src/cmd/src/bin/query_regression_runner/model.rs +++ b/src/cmd/src/bin/query_regression_runner/model.rs @@ -83,6 +83,8 @@ pub(super) struct Table { pub(super) sst_format: Option, #[serde(default = "default_show_create_engine")] pub(super) validate_show_create_engine: bool, + #[serde(default)] + pub(super) validate_timestamp_nanos: Option, } #[derive(Clone, Debug, Deserialize)] @@ -104,6 +106,8 @@ const fn default_show_create_engine() -> bool { #[derive(Clone, Debug, Deserialize)] pub(super) struct RemoteWrite { + #[serde(default)] + pub(super) input_protocol: InputProtocol, pub(super) database: String, pub(super) metric: String, pub(super) physical_table: String, @@ -122,6 +126,14 @@ pub(super) struct RemoteWrite { pub(super) read_bench: Option, } +#[derive(Clone, Debug, Deserialize, Serialize, Eq, PartialEq, Default)] +#[serde(rename_all = "snake_case")] +pub(super) enum InputProtocol { + #[default] + RemoteWrite, + OtlpMetrics, +} + #[derive(Clone, Debug, Deserialize)] pub(super) struct PromStore { pub(super) pending_rows_flush_interval: String, @@ -222,12 +234,26 @@ pub(super) struct Query { pub(super) iterations: usize, #[serde(default)] pub(super) thresholds: Map, + #[serde(default)] + pub(super) expected_cardinality: Option, + #[serde(default)] + pub(super) compare_results: bool, + #[serde(default = "default_measure")] + pub(super) measure: bool, + #[serde(default)] + pub(super) plan_contains: Option, + #[serde(default)] + pub(super) plan_absent: Option, } const fn one() -> usize { 1 } +const fn default_measure() -> bool { + true +} + #[derive(Debug, Serialize)] pub(super) struct QueryResult { pub(super) validation: Vec, diff --git a/src/cmd/src/bin/query_regression_runner/plan.rs b/src/cmd/src/bin/query_regression_runner/plan.rs index 4fcc5135cd..9d7a695e5b 100644 --- a/src/cmd/src/bin/query_regression_runner/plan.rs +++ b/src/cmd/src/bin/query_regression_runner/plan.rs @@ -20,7 +20,7 @@ use serde_json::Value; use crate::query_regression_runner::Result; use crate::query_regression_runner::model::{ - Layout, OtlpLoad, Query, RemoteWrite, Scenario, Table, + InputProtocol, Layout, OtlpLoad, Query, RemoteWrite, Scenario, Table, }; pub(super) fn load_plan(generator: &PathBuf, case_path: &PathBuf) -> Result { @@ -46,17 +46,35 @@ pub(super) fn normalize_scenario(scenario: Scenario) -> Result<(Vec, Vec< remote_write, queries, } => Ok(( - vec![Table { - database: remote_write.database, - name: remote_write.metric, - engine: "metric".to_string(), - columns: vec![], - primary_key: vec![], - time_index: None, - append_mode: None, - sst_format: None, - validate_show_create_engine: false, - }], + vec![ + if remote_write.input_protocol == InputProtocol::OtlpMetrics { + Table { + database: remote_write.database, + name: remote_write.metric, + engine: "mito".to_string(), + columns: vec![], + primary_key: vec!["host".to_string(), "instance".to_string()], + time_index: Some("greptime_timestamp".to_string()), + append_mode: None, + sst_format: None, + validate_show_create_engine: true, + validate_timestamp_nanos: Some(123), + } + } else { + Table { + database: remote_write.database, + name: remote_write.metric, + engine: "metric".to_string(), + columns: vec![], + primary_key: vec![], + time_index: None, + append_mode: None, + sst_format: None, + validate_show_create_engine: false, + validate_timestamp_nanos: None, + } + }, + ], queries, )), Scenario::OtlpTraceLoad { .. } => { diff --git a/src/cmd/src/bin/query_regression_runner/remote/ingest.rs b/src/cmd/src/bin/query_regression_runner/remote/ingest.rs index f563402afc..10ca4a7dfd 100644 --- a/src/cmd/src/bin/query_regression_runner/remote/ingest.rs +++ b/src/cmd/src/bin/query_regression_runner/remote/ingest.rs @@ -20,7 +20,7 @@ use std::time::{Duration, Instant}; use reqwest::Client; use serde_json::{Value, json}; -use crate::query_regression_runner::model::{PromStore, RemoteWrite}; +use crate::query_regression_runner::model::{InputProtocol, RemoteWrite}; use crate::query_regression_runner::plan::normalized_remote_write; use crate::query_regression_runner::sql::{ extract_count_value, http_post_sql, sql_ident, sql_string, value_f64, value_u64, @@ -29,13 +29,15 @@ use crate::query_regression_runner::{PrepareRemoteArgs, RenderRemoteConfigArgs, pub(super) async fn run_render_remote_config(args: RenderRemoteConfigArgs) -> Result<()> { let (_, remote) = normalized_remote_write(&args.fixture_generator, &args.case)?; - fs::write(args.output, frontend_prom_config(&remote.prom_store)?)?; + fs::write(args.output, frontend_remote_config(&remote)?)?; Ok(()) } -fn frontend_prom_config(prom: &PromStore) -> Result { +fn frontend_remote_config(remote: &RemoteWrite) -> Result { + let prom = &remote.prom_store; + let with_metric_engine = remote.input_protocol == InputProtocol::RemoteWrite; Ok(format!( - "[prom_store]\nenable = true\nwith_metric_engine = true\npending_rows_flush_interval = {}\nmax_batch_rows = {}\nmax_concurrent_flushes = {}\nworker_channel_capacity = {}\nmax_inflight_requests = {}\n", + "[prom_store]\nenable = true\nwith_metric_engine = {with_metric_engine}\npending_rows_flush_interval = {}\nmax_batch_rows = {}\nmax_concurrent_flushes = {}\nworker_channel_capacity = {}\nmax_inflight_requests = {}\n", serde_json::to_string(&prom.pending_rows_flush_interval)?, prom.max_batch_rows, prom.max_concurrent_flushes, @@ -71,6 +73,7 @@ pub(super) async fn run_prepare_remote(args: PrepareRemoteArgs) -> Result<()> { let report = json!({ "case_path": case_path, "scenario": "prom_remote_write_then_query", + "input_protocol": remote.input_protocol, "base": base, "candidate": candidate, "status": "ok", @@ -107,6 +110,14 @@ async fn prepare_remote_target( ) .into()); } + if remote.input_protocol == InputProtocol::OtlpMetrics { + let create = create_otlp_metric_table(client, port, remote).await; + if !create["ok"].as_bool().unwrap_or(false) { + return Err( + format!("CREATE TABLE {} failed for {name}: {create}", remote.metric).into(), + ); + } + } let (remote_write, flushes) = ingest_remote_write(generator, port, remote, client).await?; let expected_rows = remote .series_count @@ -121,16 +132,66 @@ async fn prepare_remote_target( remote.visibility_timeout_seconds, ) .await?; + let timestamp_precision = if remote.input_protocol == InputProtocol::OtlpMetrics { + verify_otlp_timestamp_precision(client, port, remote).await? + } else { + json!({"status": "skipped", "reason": "input_protocol is remote_write"}) + }; Ok(json!({ "name": name, "create_database": create_database, + "input_protocol": remote.input_protocol, + "otlp_metric_compat": if remote.input_protocol == InputProtocol::OtlpMetrics { Value::String("absent (frontend legacy-data compatibility)".to_string()) } else { Value::Null }, "remote_write": remote_write, "flushes": flushes, "visibility": visibility, + "timestamp_precision": timestamp_precision, "status": "ok", })) } +async fn create_otlp_metric_table(client: &Client, port: u16, remote: &RemoteWrite) -> Value { + http_post_sql( + client, + port, + &format!( + "CREATE TABLE {} (host STRING, instance STRING, greptime_timestamp TIMESTAMP(9) TIME INDEX, greptime_value DOUBLE, PRIMARY KEY(host, instance)) ENGINE=mito", + sql_ident(&remote.metric), + ), + &remote.database, + ) + .await +} + +async fn verify_otlp_timestamp_precision( + client: &Client, + port: u16, + remote: &RemoteWrite, +) -> Result { + let result = http_post_sql( + client, + port, + &format!( + "SELECT greptime_timestamp FROM {} ORDER BY greptime_timestamp LIMIT 1", + sql_ident(&remote.metric), + ), + &remote.database, + ) + .await; + let stored = crate::query_regression_runner::sql::extract_rows( + result.get("response").unwrap_or(&Value::Null), + ) + .first() + .and_then(|row| crate::query_regression_runner::sql::row_value(row, 0, "greptime_timestamp")) + .map(crate::query_regression_runner::sql::value_text) + .unwrap_or_default(); + let exact_remainder = stored.contains("000000123"); + if !result["ok"].as_bool().unwrap_or(false) || !exact_remainder { + return Err(format!("OTLP timestamp precision check failed: {result}").into()); + } + Ok(json!({"status": "ok", "stored_timestamp": stored, "expected_nanosecond_remainder": 123})) +} + #[derive(Clone, Debug, PartialEq, Eq)] struct SampleChunk { index: u64, @@ -302,23 +363,42 @@ fn remote_write_command( )); let mut command = vec![ generator.to_string_lossy().to_string(), - "prom-remote-write".to_string(), + match remote.input_protocol { + InputProtocol::RemoteWrite => "prom-remote-write", + InputProtocol::OtlpMetrics => "otlp-metrics", + } + .to_string(), "--endpoint".to_string(), - format!("http://127.0.0.1:{port}/v1/prometheus/write"), + match remote.input_protocol { + InputProtocol::RemoteWrite => format!("http://127.0.0.1:{port}/v1/prometheus/write"), + InputProtocol::OtlpMetrics => format!("http://127.0.0.1:{port}/v1/otlp/v1/metrics"), + }, "--database".to_string(), remote.database.clone(), "--metric".to_string(), remote.metric.clone(), - "--physical-table".to_string(), - remote.physical_table.clone(), "--series-count".to_string(), remote.series_count.to_string(), "--samples-per-series".to_string(), samples_per_series.to_string(), - "--start-unix-millis".to_string(), - start_unix_millis.to_string(), - "--step-millis".to_string(), - remote.step_millis.to_string(), + match remote.input_protocol { + InputProtocol::RemoteWrite => "--start-unix-millis", + InputProtocol::OtlpMetrics => "--start-unix-nanos", + } + .to_string(), + match remote.input_protocol { + InputProtocol::RemoteWrite => start_unix_millis.to_string(), + InputProtocol::OtlpMetrics => (start_unix_millis as u64 * 1_000_000 + 123).to_string(), + }, + match remote.input_protocol { + InputProtocol::RemoteWrite => "--step-millis", + InputProtocol::OtlpMetrics => "--step-nanos", + } + .to_string(), + match remote.input_protocol { + InputProtocol::RemoteWrite => remote.step_millis.to_string(), + InputProtocol::OtlpMetrics => (remote.step_millis as u64 * 1_000_000).to_string(), + }, "--chunk-series-count".to_string(), remote.chunk_series_count.to_string(), "--timeout-seconds".to_string(), @@ -342,13 +422,23 @@ fn remote_write_command( "--value-mixed-every".to_string(), remote.value.mixed_every.to_string(), ]; - if let Some(sample_offset) = sample_offset { + if remote.input_protocol == InputProtocol::RemoteWrite { + command.extend([ + "--physical-table".to_string(), + remote.physical_table.clone(), + ]); + } + if remote.input_protocol == InputProtocol::RemoteWrite + && let Some(sample_offset) = sample_offset + { command.extend([ "--value-sample-offset".to_string(), sample_offset.to_string(), ]); } - if let Some(total_samples_per_series) = total_samples_per_series { + if remote.input_protocol == InputProtocol::RemoteWrite + && let Some(total_samples_per_series) = total_samples_per_series + { command.extend([ "--value-total-samples-per-series".to_string(), total_samples_per_series.to_string(), @@ -364,25 +454,26 @@ async fn flush_remote_table( reason: &str, chunk_index: Option, ) -> Result { + let table = match remote.input_protocol { + InputProtocol::RemoteWrite => &remote.physical_table, + InputProtocol::OtlpMetrics => &remote.metric, + }; let mut result = http_post_sql( client, port, - &format!("ADMIN FLUSH_TABLE({})", sql_string(&remote.physical_table)), + &format!("ADMIN FLUSH_TABLE({})", sql_string(table)), &remote.database, ) .await; if !result["ok"].as_bool().unwrap_or(false) { - return Err(format!( - "ADMIN FLUSH_TABLE {} failed: {result}", - remote.physical_table - ) - .into()); + return Err(format!("ADMIN FLUSH_TABLE {} failed: {result}", table).into()); } result .as_object_mut() .ok_or("flush result must be an object")? .extend([ ("physical_table".to_string(), json!(remote.physical_table)), + ("flush_table".to_string(), json!(table)), ("reason".to_string(), json!(reason)), ("chunk_index".to_string(), json!(chunk_index)), ]); @@ -440,6 +531,7 @@ mod tests { #[test] fn schedules_remote_sample_chunks_and_flushes() { let remote = RemoteWrite { + input_protocol: InputProtocol::RemoteWrite, database: "public".to_string(), metric: "metric".to_string(), physical_table: "physical".to_string(), diff --git a/tests/perf/README.md b/tests/perf/README.md index d518fec996..c385d850c1 100644 --- a/tests/perf/README.md +++ b/tests/perf/README.md @@ -163,9 +163,9 @@ frontend path. Treat all performance conclusions as release-only; debug builds are suitable only for command wiring and correctness checks. `prepare-remote` creates the configured database if needed. The outer driver writes a per-target -frontend config enabling `[prom_store]` with metric engine storage and a non-zero -`pending_rows_flush_interval`, and validates that the logical metric table reaches -`series_count * samples_per_series` rows before trusting the query measurements. +frontend config enabling `[prom_store]` with a non-zero `pending_rows_flush_interval`; the default +`input_protocol = "remote_write"` also enables metric-engine storage. It validates that the logical +metric table reaches `series_count * samples_per_series` rows before trusting the query measurements. Use `--fixture-generator /path/to/query_perf_fixture` to provide the Rust helper to the outer driver. @@ -448,3 +448,29 @@ Additional SQL optimizer cases: `LIMIT`. - `sql_join_filter_order`: two direct-SST tables joined on a shared tag with time filters, aggregate ordering, and `LIMIT`. + +### Native OTLP nanosecond LastRow case + +`otlp_ns_last_row_9070` reuses the remote-write-then-query lifecycle with +`input_protocol = "otlp_metrics"`. Before each fresh target is loaded, +`prepare-remote` creates its lowercase `ENGINE=mito` metric table with +`greptime_timestamp TIMESTAMP(9)`, then sends binary OTLP gauge protobufs to +`/v1/otlp/v1/metrics`. The load report records `input_protocol`, HTTP statuses, +and decoded OTLP `partial_success.rejected_data_points`; a nonzero rejection +fails preparation. It flushes the logical metric table, checks all 262144 rows, +and verifies the stored timestamp retains the `+123ns` remainder. + +The measured requests are ordinary `TQL EVAL` instant and range selectors (3 +warmups, 9 iterations). Untimed `TQL EXPLAIN VERBOSE` evidence is retained in +the report, and every measured response is row-normalized and compared with +the base target, ignoring only timing metadata. This is explicitly +legacy-data query coverage; it does not assert the new default OTLP +millisecond table behavior. Use the usual remote-case dispatch: + +```bash +uv run --no-project python .github/scripts/query-regression-run.py \ + --cases tests/perf/query_cases/otlp_ns_last_row_9070/case.toml \ + --base-bin /path/to/base/greptime --candidate-bin /path/to/candidate/greptime \ + --fixture-generator /path/to/query_perf_fixture \ + --runner /path/to/query_regression_runner --work-dir /path/to/work +``` diff --git a/tests/perf/query_cases/otlp_ns_last_row_9070/case.toml b/tests/perf/query_cases/otlp_ns_last_row_9070/case.toml new file mode 100644 index 0000000000..ee1daf6a12 --- /dev/null +++ b/tests/perf/query_cases/otlp_ns_last_row_9070/case.toml @@ -0,0 +1,67 @@ +# Native OTLP metrics regression coverage for #9070. The table is explicitly +# pre-created so the native OTLP path preserves nanosecond timestamps rather +# than selecting a legacy compatibility schema. + +[case] +name = "otlp_ns_last_row_9070" +description = "Native OTLP gauge ingestion with nanosecond timestamps and ordinary PromQL LastRow queries" +issue = "https://github.com/GreptimeTeam/greptimedb/issues/9070" + +[scenario] +kind = "prom_remote_write_then_query" + +[scenario.remote_write] +input_protocol = "otlp_metrics" +database = "public" +metric = "otlp_ns_last_row_9070" +physical_table = "greptime_physical_table" +series_count = 256 +samples_per_series = 1024 +start_unix_millis = 1_704_067_200_000 +step_millis = 100 +chunk_series_count = 256 +timeout_seconds = 300 +visibility_timeout_seconds = 120 + +# Legacy-data query coverage only: this case intentionally does not test the +# new default OTLP millisecond table behavior. + +[[scenario.queries]] +name = "instant_selector_last_row_candidate" +kind = "tql" +query = "TQL EVAL (1704067303, 1704067303, '1s') otlp_ns_last_row_9070{host=~'host.*'}" +warmup = 3 +iterations = 9 +compare_results = true +expected_cardinality = 256 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 25 + +[[scenario.queries]] +name = "range_selector_last_row_control" +kind = "tql" +query = "TQL EVAL (1704067293, 1704067303, '1s') otlp_ns_last_row_9070{host=~'host.*'}" +warmup = 3 +iterations = 9 +compare_results = true +expected_cardinality = 2816 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 25 + +# Keep plan evidence untimed. It is reported with validation and never enters +# the latency sample set. +[[scenario.queries]] +name = "instant_last_row_plan" +kind = "tql" +query = "TQL EXPLAIN VERBOSE (1704067303, 1704067303, '1s') otlp_ns_last_row_9070{host=~'host.*'}" +measure = false +plan_contains = "LastRow" + +[[scenario.queries]] +name = "range_no_last_row_plan" +kind = "tql" +query = "TQL EXPLAIN VERBOSE (1704067293, 1704067303, '1s') otlp_ns_last_row_9070{host=~'host.*'}" +measure = false +plan_absent = "LastRow"