From 4b020fe5b34da4c766e4ab9cd2278f9c8e310e2f Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Thu, 10 Sep 2026 17:45:32 +0800 Subject: [PATCH] test: compare complete OTLP query schemas and row values Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- .../bin/query_regression_runner/measure.rs | 393 +++++++++++++++--- .../src/bin/query_regression_runner/sql.rs | 43 +- tests/perf/README.md | 8 +- .../otlp_ns_last_row_9070/case.toml | 6 +- 4 files changed, 390 insertions(+), 60 deletions(-) diff --git a/src/cmd/src/bin/query_regression_runner/measure.rs b/src/cmd/src/bin/query_regression_runner/measure.rs index 9f2bdc6975..c92d8ed601 100644 --- a/src/cmd/src/bin/query_regression_runner/measure.rs +++ b/src/cmd/src/bin/query_regression_runner/measure.rs @@ -21,7 +21,9 @@ 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_prom_range_query, http_post_sql, sql_ident}; +use crate::query_regression_runner::sql::{ + http_post_prom_range_query, http_post_sql, http_post_sql_with_schema, sql_ident, +}; use crate::query_regression_runner::{MeasureArgs, Result}; pub(super) async fn run_measure(args: MeasureArgs) -> Result<()> { @@ -131,6 +133,8 @@ async fn post_query(client: &Client, port: u16, query: &Query, db: &str) -> Valu db, ) .await + } else if query.compare_results || query.expected_cardinality.is_some() { + http_post_sql_with_schema(client, port, &query.query, db).await } else { http_post_sql(client, port, &query.query, db).await } @@ -272,15 +276,28 @@ fn compare_measured_results( base: &mut QueryResult, candidate: &mut QueryResult, ) { + let mut errors = Vec::new(); for query in queries { if !query.measure || (!query.compare_results && query.expected_cardinality.is_none()) { continue; } + if query.compare_results && !matches!(query.kind.as_deref(), Some("sql" | "tql")) { + errors.push(json!({ + "query": query.name, + "error": "result comparison only supports sql and tql queries", + })); + continue; + } + let Some(base_measurement) = base .measurements .iter() .find(|measurement| measurement.name == query.name) else { + errors.push(json!({ + "query": query.name, + "error": "missing base measurement", + })); continue; }; let Some(candidate_measurement) = candidate @@ -288,71 +305,162 @@ fn compare_measured_results( .iter() .find(|measurement| measurement.name == query.name) else { + errors.push(json!({ + "query": query.name, + "error": "missing candidate measurement", + })); 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), + + if let Some(expected) = query.expected_cardinality { + validate_cardinality( + query, + "base", + &base_measurement.samples, + expected, + &mut errors, ); - let candidate_rows = crate::query_regression_runner::sql::extract_rows( - candidate_sample.get("response").unwrap_or(&Value::Null), + validate_cardinality( + query, + "candidate", + &candidate_measurement.samples, + expected, + &mut errors, ); - 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 !query.compare_results { + continue; + } + if base_measurement.samples.len() != candidate_measurement.samples.len() { + errors.push(json!({ + "query": query.name, + "error": format!( + "sample count differs: base={} candidate={}", + base_measurement.samples.len(), + candidate_measurement.samples.len(), + ), + })); + continue; + } + + for iteration in 0..base_measurement.samples.len() { + let base_records = extract_records( + base_measurement.samples[iteration] + .get("response") + .unwrap_or(&Value::Null), + ); + let candidate_records = extract_records( + candidate_measurement.samples[iteration] + .get("response") + .unwrap_or(&Value::Null), + ); + match (base_records, candidate_records) { + (Ok(base_records), Ok(candidate_records)) => { + if !records_match(base_records, candidate_records) { + errors.push(json!({ + "query": query.name, + "iteration": iteration, + "error": "base/candidate response records differ", + })); + } + } + (Err(error), _) => errors.push(json!({ + "query": query.name, + "iteration": iteration, + "error": format!("malformed base records response: {error}"), + })), + (_, Err(error)) => errors.push(json!({ + "query": query.name, + "iteration": iteration, + "error": format!("malformed candidate records response: {error}"), + })), } } } + candidate.validation_errors.extend(errors); if !candidate.validation_errors.is_empty() { candidate.status = "failed".to_string(); } } -fn normalized_rows(rows: Vec) -> Vec { +struct Records<'a> { + schema: &'a Value, + rows: &'a [Value], +} + +fn extract_records(response: &Value) -> std::result::Result, &'static str> { + let output = response + .as_object() + .and_then(|response| response.get("output")) + .and_then(Value::as_array) + .ok_or("response has no output array")?; + let [output] = output.as_slice() else { + return Err("response output must contain exactly one records result"); + }; + let records = output + .as_object() + .and_then(|output| output.get("records")) + .and_then(Value::as_object) + .ok_or("output has no records object")?; + let schema = records + .get("schema") + .filter(|schema| schema.is_object()) + .ok_or("records has no schema object")?; + if schema + .get("column_schemas") + .and_then(Value::as_array) + .is_none() + { + return Err("schema has no column_schemas array"); + } + let rows = records + .get("rows") + .and_then(Value::as_array) + .ok_or("records has no rows array")?; + Ok(Records { schema, rows }) +} + +fn validate_cardinality( + query: &Query, + target: &str, + samples: &[Value], + expected: u64, + errors: &mut Vec, +) { + for (iteration, sample) in samples.iter().enumerate() { + match extract_records(sample.get("response").unwrap_or(&Value::Null)) { + Ok(records) if records.rows.len() as u64 == expected => {} + Ok(records) => errors.push(json!({ + "query": query.name, + "iteration": iteration, + "error": format!( + "expected cardinality {expected}, got {target}={}", + records.rows.len(), + ), + })), + Err(error) => errors.push(json!({ + "query": query.name, + "iteration": iteration, + "error": format!("malformed {target} records response: {error}"), + })), + } + } +} + +fn records_match(base: Records<'_>, candidate: Records<'_>) -> bool { + base.schema == candidate.schema && row_multiset(base.rows) == row_multiset(candidate.rows) +} + +fn row_multiset(rows: &[Value]) -> Vec { let mut rows = rows - .into_iter() - .map(|mut row| { - strip_timing_metadata(&mut row); - serde_json::to_string(&row).unwrap_or_default() - }) + .iter() + .map(|row| serde_json::to_string(row).expect("JSON values are serializable")) .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") @@ -553,6 +661,195 @@ mod tests { ); } + fn compared_query(iterations: usize, expected_cardinality: Option) -> Query { + Query { + name: Some("q".to_string()), + kind: Some("tql".to_string()), + query: "TQL EVAL (0, 1, '1s') 1".to_string(), + start: None, + end: None, + step: None, + warmup: 0, + iterations, + expected_cardinality, + compare_results: true, + measure: true, + plan_contains: None, + plan_absent: None, + thresholds: Map::new(), + } + } + + fn records_sample(column: &str, data_type: &str, rows: Value) -> Value { + json!({ + "response": { + "output": [{ + "records": { + "schema": {"column_schemas": [{"name": column, "data_type": data_type}]}, + "rows": rows, + } + }], + "execution_time_ms": 10, + "elapsed": 20, + "cost": 30, + } + }) + } + + fn compared_results(samples: Vec) -> QueryResult { + QueryResult { + validation: vec![], + validation_errors: vec![], + measurements: vec![Measurement { + name: Some("q".to_string()), + kind: Some("tql".to_string()), + iterations: samples.len(), + samples, + latency_ms_median: None, + latency_ms_p95: None, + status: "ok".to_string(), + }], + status: "ok".to_string(), + } + } + + fn compare_samples( + expected_cardinality: Option, + base_samples: Vec, + candidate_samples: Vec, + ) -> QueryResult { + let query = compared_query(base_samples.len(), expected_cardinality); + let mut base = compared_results(base_samples); + let mut candidate = compared_results(candidate_samples); + compare_measured_results(&[query], &mut base, &mut candidate); + candidate + } + + #[test] + fn comparison_rejects_changed_column_name() { + let candidate = compare_samples( + None, + vec![records_sample("value", "Int64", json!([[1]]))], + vec![records_sample("other", "Int64", json!([[1]]))], + ); + assert_eq!(candidate.validation_errors.len(), 1); + assert_eq!(candidate.validation_errors[0]["iteration"], 0); + } + + #[test] + fn comparison_rejects_changed_column_type() { + let candidate = compare_samples( + None, + vec![records_sample("value", "Int64", json!([[1]]))], + vec![records_sample("value", "String", json!([[1]]))], + ); + assert_eq!(candidate.validation_errors.len(), 1); + } + + #[test] + fn comparison_preserves_cost_and_elapsed_in_row_json() { + let base = records_sample("value", "Json", json!([{"cost": 1, "elapsed": 2}])); + let candidate = records_sample("value", "Json", json!([{"cost": 3, "elapsed": 4}])); + let candidate = compare_samples(None, vec![base], vec![candidate]); + assert_eq!(candidate.validation_errors.len(), 1); + } + + #[test] + fn comparison_ignores_response_execution_timing() { + let base = records_sample("value", "Int64", json!([[1]])); + let mut candidate = records_sample("value", "Int64", json!([[1]])); + candidate["response"]["execution_time_ms"] = json!(99); + candidate["response"]["elapsed"] = json!(88); + candidate["response"]["cost"] = json!(77); + let candidate = compare_samples(None, vec![base], vec![candidate]); + assert!(candidate.validation_errors.is_empty()); + } + + #[test] + fn comparison_accepts_unordered_equal_rows() { + let candidate = compare_samples( + None, + vec![records_sample("value", "Int64", json!([[1], [2]]))], + vec![records_sample("value", "Int64", json!([[2], [1]]))], + ); + assert!(candidate.validation_errors.is_empty()); + } + + #[test] + fn comparison_rejects_duplicate_row_mismatch() { + let candidate = compare_samples( + None, + vec![records_sample("value", "Int64", json!([[1], [1]]))], + vec![records_sample("value", "Int64", json!([[1], [2]]))], + ); + assert_eq!(candidate.validation_errors.len(), 1); + } + + #[test] + fn comparison_rejects_cardinality_mismatch() { + let sample = records_sample("value", "Int64", json!([[1]])); + let candidate = compare_samples(Some(2), vec![sample.clone()], vec![sample]); + assert_eq!(candidate.validation_errors.len(), 2); + assert!(candidate.validation_errors.iter().all(|error| { + error["error"] + .as_str() + .unwrap() + .contains("expected cardinality 2") + })); + } + + #[test] + fn comparison_rejects_later_iteration_mismatch() { + let candidate = compare_samples( + None, + vec![ + records_sample("value", "Int64", json!([[1]])), + records_sample("value", "Int64", json!([[2]])), + ], + vec![ + records_sample("value", "Int64", json!([[1]])), + records_sample("value", "Int64", json!([[3]])), + ], + ); + assert_eq!(candidate.validation_errors.len(), 1); + assert_eq!(candidate.validation_errors[0]["iteration"], 1); + } + + #[test] + fn comparison_rejects_missing_iteration() { + let candidate = compare_samples( + None, + vec![ + records_sample("value", "Int64", json!([[1]])), + records_sample("value", "Int64", json!([[2]])), + ], + vec![records_sample("value", "Int64", json!([[1]]))], + ); + assert_eq!(candidate.validation_errors.len(), 1); + assert!( + candidate.validation_errors[0]["error"] + .as_str() + .unwrap() + .contains("sample count differs: base=2 candidate=1") + ); + } + + #[test] + fn comparison_rejects_malformed_records_response() { + let candidate = compare_samples( + None, + vec![json!({"response": {"output": []}})], + vec![json!({"response": {"output": []}})], + ); + assert_eq!(candidate.validation_errors.len(), 1); + assert!( + candidate.validation_errors[0]["error"] + .as_str() + .unwrap() + .contains("malformed base records response") + ); + } + #[test] fn threshold_rejects_unknown_keys_and_zero_base() { let query = Query { diff --git a/src/cmd/src/bin/query_regression_runner/sql.rs b/src/cmd/src/bin/query_regression_runner/sql.rs index b33c0a1c22..ed9eef3688 100644 --- a/src/cmd/src/bin/query_regression_runner/sql.rs +++ b/src/cmd/src/bin/query_regression_runner/sql.rs @@ -117,12 +117,27 @@ 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; + http_post_sql_with_format(client, port, sql, db, "json").await +} + +pub(super) async fn http_post_sql_with_schema( + client: &Client, + port: u16, + sql: &str, + db: &str, +) -> Value { + http_post_sql_with_format(client, port, sql, db, "greptimedb_v1").await +} + +async fn http_post_sql_with_format( + client: &Client, + port: u16, + sql: &str, + db: &str, + format: &str, +) -> Value { + let form = sql_form(sql, db, format); + let mut sample = post_form(client, format!("http://127.0.0.1:{port}/v1/sql"), &form).await; sample .as_object_mut() .expect("HTTP samples are objects") @@ -130,6 +145,10 @@ pub(super) async fn http_post_sql(client: &Client, port: u16, sql: &str, db: &st sample } +fn sql_form<'a>(sql: &'a str, db: &'a str, format: &'a str) -> [(&'static str, &'a str); 3] { + [("sql", sql), ("db", db), ("format", format)] +} + /// 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`) @@ -268,6 +287,18 @@ mod tests { ); } + #[test] + fn sql_form_uses_requested_response_format() { + assert_eq!( + sql_form("SELECT 1", "public", "greptimedb_v1"), + [ + ("sql", "SELECT 1"), + ("db", "public"), + ("format", "greptimedb_v1"), + ] + ); + } + #[test] fn top_level_errors_do_not_inspect_rows() { assert!(response_has_error(&json!({"error_code": 7}))); diff --git a/tests/perf/README.md b/tests/perf/README.md index c385d850c1..31b0a1df27 100644 --- a/tests/perf/README.md +++ b/tests/perf/README.md @@ -461,9 +461,11 @@ 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 +warmups, 9 iterations). Untimed `TQL ANALYZE VERBOSE` evidence is retained in +the report. Measured queries request `greptimedb_v1` responses and compare the +complete ordered column schema and unordered row multiset with the base target +on every iteration. Row values are preserved verbatim; response-envelope timing +metadata is outside the comparison. This is explicitly legacy-data query coverage; it does not assert the new default OTLP millisecond table behavior. Use the usual remote-case dispatch: 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 index ee1daf6a12..d74c53b3c8 100644 --- a/tests/perf/query_cases/otlp_ns_last_row_9070/case.toml +++ b/tests/perf/query_cases/otlp_ns_last_row_9070/case.toml @@ -55,13 +55,13 @@ max_candidate_latency_regression_pct = 25 [[scenario.queries]] name = "instant_last_row_plan" kind = "tql" -query = "TQL EXPLAIN VERBOSE (1704067303, 1704067303, '1s') otlp_ns_last_row_9070{host=~'host.*'}" +query = "TQL ANALYZE VERBOSE (1704067303, 1704067303, '1s') otlp_ns_last_row_9070{host=~'host.*'}" measure = false -plan_contains = "LastRow" +plan_contains = "LastRow { after_merge: true }" [[scenario.queries]] name = "range_no_last_row_plan" kind = "tql" -query = "TQL EXPLAIN VERBOSE (1704067293, 1704067303, '1s') otlp_ns_last_row_9070{host=~'host.*'}" +query = "TQL ANALYZE VERBOSE (1704067293, 1704067303, '1s') otlp_ns_last_row_9070{host=~'host.*'}" measure = false plan_absent = "LastRow"