diff --git a/src/cmd/src/bin/query_regression_runner/measure.rs b/src/cmd/src/bin/query_regression_runner/measure.rs index fd80813f40e..290289907e0 100644 --- a/src/cmd/src/bin/query_regression_runner/measure.rs +++ b/src/cmd/src/bin/query_regression_runner/measure.rs @@ -13,6 +13,8 @@ // limitations under the License. use std::collections::HashMap; +use std::io::{BufWriter, Write}; +use std::path::{Path, PathBuf}; use std::time::Duration; use std::{fs, io}; @@ -49,14 +51,25 @@ 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 response_artifacts = args.output.as_deref().map(ResponseArtifacts::new); + let base = run_target( + args.base_http_port, + &tables, + &configured_queries, + &client, + 0, + response_artifacts.as_ref(), + ) + .await?; let candidate = run_target( args.candidate_http_port, &tables, &configured_queries, &client, + 1, + response_artifacts.as_ref(), ) - .await; + .await?; let thresholds = enforce_thresholds(&configured_queries, &base, &candidate)?; let status = if base.status == "failed" || candidate.status == "failed" @@ -68,7 +81,7 @@ pub(super) async fn run_measure(args: MeasureArgs) -> Result<()> { } else { "ok" }; - let report = json!({ + let mut report = json!({ "case_path": case_path, "case": case_metadata, "scenario": scenario_value, @@ -79,11 +92,11 @@ pub(super) async fn run_measure(args: MeasureArgs) -> Result<()> { "thresholds": thresholds, "status": status, }); - let text = format!("{}\n", serde_json::to_string_pretty(&report)?); if let Some(path) = args.output { - fs::write(path, &text)?; + report["report_format_version"] = Value::from(2); + write_report(&path, &report)?; } - print!("{text}"); + write_stdout_report(&report)?; if status == "failed" { std::process::exit(1); } @@ -131,7 +144,9 @@ async fn run_target( tables: &[Table], configured_queries: &[Query], client: &Client, -) -> QueryResult { + target_index: usize, + response_artifacts: Option<&ResponseArtifacts>, +) -> Result { let mut queries = configured_queries.to_vec(); if queries.is_empty() { queries.push(Query { @@ -150,52 +165,79 @@ async fn run_target( let db = &tables[0].database; let mut validation = Vec::new(); let mut validation_errors = Vec::new(); - for table in tables { + for (table_index, table) in tables.iter().enumerate() { let sql = format!("SHOW CREATE TABLE {}", sql_ident(&table.name)); - let sample = http_post_sql(client, port, &sql, &table.database).await; + let mut sample = http_post_sql(client, port, &sql, &table.database).await; + let validation_index = validation.len(); + let validation_error_start = validation_errors.len(); if !sample["ok"].as_bool().unwrap_or(false) { - validation_errors.push(json!({ - "sql": sql, - "error": sample.get("error"), - "response": sample.get("response"), - })); + validation_errors.push(validation_error( + &sql, + None, + sample.get("error").cloned().unwrap_or(Value::Null), + validation_index, + )); } else { for error in validate_show_create(&sample, table) { - validation_errors.push(json!({ - "sql": sql, - "error": error, - "response": sample.get("response"), - })); + validation_errors.push(validation_error( + &sql, + None, + Value::String(error.to_string()), + validation_index, + )); } } + persist_response( + response_artifacts, + &mut sample, + target_index, + table_index, + 0, + 0, + )?; + attach_response(&sample, &mut validation_errors[validation_error_start..]); validation.push(sample); } - let first = post_query(client, port, &queries[0], db).await; + let mut first = post_query(client, port, &queries[0], db).await; + let validation_index = validation.len(); + let validation_error_start = validation_errors.len(); 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_errors.push(validation_error( + &queries[0].query, + None, + first.get("error").cloned().unwrap_or(Value::Null), + validation_index, + )); } + persist_response(response_artifacts, &mut first, target_index, 0, 1, 0)?; + attach_response(&first, &mut validation_errors[validation_error_start..]); validation.push(first); let mut measurements = Vec::with_capacity(queries.len()); - for query in &queries { - for _ in 0..query.warmup { - let warmup = post_query(client, port, query, db).await; + for (query_index, query) in queries.iter().enumerate() { + for warmup_index in 0..query.warmup { + let mut warmup = post_query(client, port, query, db).await; if !warmup["ok"].as_bool().unwrap_or(false) { - validation_errors.push(json!({ + let mut failure = json!({ "sql": query.query, "phase": "warmup", - "error": warmup.get("error"), - "response": warmup.get("response"), - })); + "error": warmup.get("error").cloned(), + }); + move_response(&mut warmup, &mut failure); + persist_response( + response_artifacts, + &mut failure, + target_index, + query_index, + 2, + warmup_index, + )?; + validation_errors.push(failure); } } let mut samples = Vec::with_capacity(query.iterations); let mut good_latencies = Vec::with_capacity(query.iterations); - for _ in 0..query.iterations { + for sample_index in 0..query.iterations { let mut sample = post_query(client, port, query, db).await; let execution_time = sample .get("response") @@ -211,6 +253,14 @@ async fn run_target( if sample["ok"].as_bool().unwrap_or(false) { good_latencies.push(sample["latency_ms"].as_f64().unwrap_or_default()); } + persist_response( + response_artifacts, + &mut sample, + target_index, + query_index, + 3, + sample_index, + )?; samples.push(sample); } let median = (!good_latencies.is_empty()).then(|| median(&good_latencies)); @@ -231,11 +281,144 @@ async fn run_target( }); } let failed = !validation_errors.is_empty() || measurements.iter().any(|m| m.status == "failed"); - QueryResult { + Ok(QueryResult { validation, validation_errors, measurements, status: if failed { "failed" } else { "ok" }.to_string(), + }) +} + +fn write_report(path: &Path, report: &Value) -> Result<()> { + let file = fs::File::create(path)?; + let mut writer = BufWriter::new(file); + serde_json::to_writer_pretty(&mut writer, report)?; + writer.write_all(b"\n")?; + writer.flush()?; + Ok(()) +} + +fn write_stdout_report(report: &Value) -> Result<()> { + let stdout = io::stdout(); + let mut writer = stdout.lock(); + serde_json::to_writer_pretty(&mut writer, report)?; + writer.write_all(b"\n")?; + writer.flush()?; + Ok(()) +} + +struct ResponseArtifacts { + report_dir: PathBuf, +} + +impl ResponseArtifacts { + fn new(report_path: &Path) -> Self { + Self { + report_dir: report_path + .parent() + .unwrap_or_else(|| Path::new(".")) + .to_path_buf(), + } + } + + fn persist( + &self, + sample: &mut Value, + target: usize, + query: usize, + phase: usize, + sample_index: usize, + ) -> Result<()> { + let Some(response) = sample.get("response") else { + return Ok(()); + }; + let relative_path = PathBuf::from("logs") + .join("responses") + .join(target.to_string()) + .join(query.to_string()) + .join(phase.to_string()) + .join(format!("{sample_index}.json")); + let path = self.report_dir.join(&relative_path); + let parent = path + .parent() + .ok_or("response artifact path has no parent")?; + fs::create_dir_all(parent)?; + let file = fs::File::create(path)?; + let mut writer = BufWriter::new(file); + serde_json::to_writer(&mut writer, response)?; + writer.write_all(b"\n")?; + writer.flush()?; + + let sample = sample + .as_object_mut() + .ok_or("HTTP sample is not an object")?; + sample.remove("response"); + sample.insert( + "response_ref".to_string(), + json!({"path": relative_path, "format": "json"}), + ); + Ok(()) + } +} + +fn persist_response( + response_artifacts: Option<&ResponseArtifacts>, + sample: &mut Value, + target: usize, + query: usize, + phase: usize, + sample_index: usize, +) -> Result<()> { + if let Some(response_artifacts) = response_artifacts { + response_artifacts.persist(sample, target, query, phase, sample_index)?; + } + Ok(()) +} + +fn attach_response(sample: &Value, errors: &mut [Value]) { + let (field, response) = if let Some(response_ref) = sample.get("response_ref") { + ("response_ref", response_ref) + } else if let Some(response) = sample.get("response") { + ("response", response) + } else { + return; + }; + for error in errors { + error + .as_object_mut() + .expect("validation errors are objects") + .insert(field.to_string(), response.clone()); + } +} + +fn validation_error( + sql: &str, + phase: Option<&str>, + error: Value, + validation_sample: usize, +) -> Value { + let mut error_record = json!({ + "sql": sql, + "error": error, + "validation_sample": validation_sample, + }); + if let Some(phase) = phase { + error_record + .as_object_mut() + .expect("validation errors are objects") + .insert("phase".to_string(), Value::String(phase.to_string())); + } + error_record +} + +fn move_response(from: &mut Value, to: &mut Value) { + if let Some(response) = from + .as_object_mut() + .and_then(|sample| sample.remove("response")) + { + to.as_object_mut() + .expect("validation errors are objects") + .insert("response".to_string(), response); } } @@ -389,6 +572,51 @@ fn value_as_f64(value: &Value) -> Result { mod tests { use super::*; + #[test] + fn persists_response_as_report_relative_reference() { + let tempdir = tempfile::tempdir().unwrap(); + let report = tempdir.path().join("query-regression-report.json"); + let artifacts = ResponseArtifacts::new(&report); + let mut sample = json!({"ok": true, "response": {"data": ["full response"]}}); + + artifacts.persist(&mut sample, 1, 2, 3, 4).unwrap(); + + assert!(sample.get("response").is_none()); + assert_eq!( + sample["response_ref"], + json!({"path": "logs/responses/1/2/3/4.json", "format": "json"}) + ); + assert_eq!( + fs::read_to_string(tempdir.path().join("logs/responses/1/2/3/4.json")).unwrap(), + "{\"data\":[\"full response\"]}\n" + ); + } + + #[test] + fn failed_response_persistence_keeps_body_without_reference() { + let tempdir = tempfile::tempdir().unwrap(); + let report_parent = tempdir.path().join("not-a-directory"); + fs::write(&report_parent, "not a directory").unwrap(); + let artifacts = ResponseArtifacts::new(&report_parent.join("report.json")); + let mut sample = json!({"ok": true, "response": {"data": ["full response"]}}); + + assert!(artifacts.persist(&mut sample, 0, 0, 0, 0).is_err()); + assert_eq!(sample["response"], json!({"data": ["full response"]})); + assert!(sample.get("response_ref").is_none()); + } + + #[test] + fn bodyless_sample_has_no_response_reference() { + let tempdir = tempfile::tempdir().unwrap(); + let artifacts = + ResponseArtifacts::new(&tempdir.path().join("query-regression-report.json")); + let mut sample = json!({"ok": false, "error": "connection refused"}); + + artifacts.persist(&mut sample, 0, 0, 0, 0).unwrap(); + + assert!(sample.get("response_ref").is_none()); + } + #[test] fn median_and_p95_match_python() { assert_eq!(median(&[1.0, 8.0, 3.0, 4.0]), 3.5); diff --git a/src/cmd/src/bin/query_regression_runner/sql.rs b/src/cmd/src/bin/query_regression_runner/sql.rs index b33c0a1c22b..e7357689f1c 100644 --- a/src/cmd/src/bin/query_regression_runner/sql.rs +++ b/src/cmd/src/bin/query_regression_runner/sql.rs @@ -15,7 +15,7 @@ use std::time::Instant; use reqwest::Client; -use serde_json::{Value, json}; +use serde_json::{Map, Value, json}; use crate::query_regression_runner::Result; @@ -180,19 +180,20 @@ async fn post_form(client: &Client, url: String, form: &[(&str, &str)]) -> Value Ok(raw) => { let body = serde_json::from_str(&raw).unwrap_or_else(|_| json!({"raw": raw})); let ok = status < 400 && !response_has_error(&body); - let mut sample = json!({ - "ok": ok, - "status": status, - "latency_ms": started.elapsed().as_secs_f64() * 1000.0, - "response": body, - }); + let mut sample = Map::new(); + sample.insert("ok".to_string(), Value::Bool(ok)); + sample.insert("status".to_string(), Value::from(status)); + // Capture latency only after the entire body has been read, + // parsed, and classified, before any report persistence. + sample.insert( + "latency_ms".to_string(), + Value::from(started.elapsed().as_secs_f64() * 1000.0), + ); + sample.insert("response".to_string(), body); if status >= 400 { - sample - .as_object_mut() - .expect("HTTP samples are objects") - .insert("error".to_string(), Value::String(format!("HTTP {status}"))); + sample.insert("error".to_string(), Value::String(format!("HTTP {status}"))); } - sample + Value::Object(sample) } Err(error) => json!({ "ok": false, diff --git a/tests/perf/README.md b/tests/perf/README.md index a9e7575ae99..25f37630ded 100644 --- a/tests/perf/README.md +++ b/tests/perf/README.md @@ -351,7 +351,11 @@ with `render-remote-config`, starts both clusters, runs `prepare-remote` and `measure`, stops both datanodes, then runs `finalize-remote`. Finalization owns storage inspection and read-bench against the quiescent data homes. Both paths write `query-regression-report.json` in the case work directory and always clean -up the remaining components. +up the remaining components. When `query_regression_runner measure` receives `--output`, its +report has `report_format_version: 2`; every response body otherwise exposed by the +report is written as JSON below report-relative `logs/responses////.json` +and replaced with `response_ref` metadata (`path` and `format`). The workflow already uploads +that `logs/**` tree. Without `--output`, stdout preserves the legacy inline response bodies. For local orchestration after building the three binaries, invoke the outer driver directly: