From ec9d85b32b8f0723df008ed08e099ca96835cf01 Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Tue, 29 Sep 2026 15:50:18 +0800 Subject: [PATCH] test(perf): measure fulltext insert and flush batches with native SQL response Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/cmd/src/bin/query_perf_fixture/case.rs | 99 +++++++++ src/cmd/src/bin/query_perf_fixture/mod.rs | 1 + .../bin/query_regression_runner/measure.rs | 135 +++++++++++- .../src/bin/query_regression_runner/model.rs | 65 ++++++ .../src/bin/query_regression_runner/plan.rs | 67 +++++- .../src/bin/query_regression_runner/sql.rs | 195 +++++++++++++++++- tests/perf/README.md | 25 +++ .../opt02_fulltext_flush/case.toml | 149 +++++++++++++ 8 files changed, 721 insertions(+), 15 deletions(-) create mode 100644 tests/perf/query_cases/opt02_fulltext_flush/case.toml diff --git a/src/cmd/src/bin/query_perf_fixture/case.rs b/src/cmd/src/bin/query_perf_fixture/case.rs index 271c3928e35..43f10549cc8 100644 --- a/src/cmd/src/bin/query_perf_fixture/case.rs +++ b/src/cmd/src/bin/query_perf_fixture/case.rs @@ -443,6 +443,8 @@ pub(super) fn default_parallelism() -> u64 { #[cfg(test)] mod tests { + use serde_json::json; + use super::*; #[test] @@ -496,6 +498,61 @@ candidate_setup_sql = ["CREATE TABLE candidate_table", "ALTER TABLE candidate_ta ] ); } + + #[test] + fn queries_preserve_response_format_in_the_normalized_plan() { + let case: CaseFile = toml::from_str( + r#" +[scenario] +kind = "prom_remote_write_then_query" + +[scenario.remote_write] +metric = "metric" + +[[scenario.queries]] +name = "insert_flush" +kind = "sql" +query = "INSERT INTO t (ts, msg) SELECT ts, msg FROM src; ADMIN FLUSH_TABLE('t')" +warmup = 2 +iterations = 9 +response_format = "greptimedb_v1" + +[[scenario.queries]] +name = "count" +kind = "sql" +query = "SELECT count(*) FROM t" +"#, + ) + .unwrap(); + let plan = serde_json::to_value(&case.scenario).unwrap(); + assert_eq!( + plan["queries"][0]["query"], + "INSERT INTO t (ts, msg) SELECT ts, msg FROM src; ADMIN FLUSH_TABLE('t')" + ); + assert_eq!(plan["queries"][0]["response_format"], "greptimedb_v1"); + assert_eq!(plan["queries"][0]["warmup"], 2); + assert_eq!(plan["queries"][0]["iterations"], 9); + // Queries that do not opt in keep the historical plan shape. + assert!(plan["queries"][1].get("response_format").is_none()); + + // Minimal plan-time validation: invalid values fail before any request. + for value in [json!("csv"), json!(5), json!(true)] { + assert!( + validate_queries(&[ + json!({"kind": "sql", "query": "SELECT 1", "response_format": value}) + ]) + .is_err() + ); + } + assert!( + validate_queries(&[ + json!({"kind": "prom_http", "query": "x", "response_format": "json"}) + ]) + .is_err() + ); + validate_queries(&[json!({"kind": "sql", "query": "SELECT 1", "response_format": "json"})]) + .unwrap(); + } } impl Scenario { @@ -507,6 +564,14 @@ impl Scenario { } } + pub(super) fn queries(&self) -> &[serde_json::Value] { + match self { + Scenario::DirectReadableSst(scenario) => &scenario.queries, + Scenario::PromRemoteWriteThenQuery(scenario) => &scenario.queries, + Scenario::OtlpTraceLoad(_) => &[], + } + } + pub(super) fn direct_readable_sst(&self) -> &DirectReadableSstScenario { match self { Scenario::DirectReadableSst(scenario) => scenario, @@ -515,6 +580,40 @@ impl Scenario { } } +/// Response formats the query-regression runner accepts for +/// `[[scenario.queries]].response_format`. +pub(super) const QUERY_RESPONSE_FORMATS: [&str; 2] = ["json", "greptimedb_v1"]; + +/// Minimal validation of a query field the runner consumes; queries stay +/// `serde_json::Value`, so this only rejects what the runner cannot honor. +pub(super) fn validate_queries(queries: &[serde_json::Value]) -> Result<(), String> { + for (index, query) in queries.iter().enumerate() { + let Some(object) = query.as_object() else { + return Err(format!("scenario.queries[{index}] must be a table")); + }; + let Some(format) = object.get("response_format") else { + continue; + }; + let Some(format) = format.as_str() else { + return Err(format!( + "scenario.queries[{index}].response_format must be a string" + )); + }; + if !QUERY_RESPONSE_FORMATS.contains(&format) { + return Err(format!( + "scenario.queries[{index}].response_format must be one of {}", + QUERY_RESPONSE_FORMATS.join(", ") + )); + } + if object.get("kind").and_then(serde_json::Value::as_str) == Some("prom_http") { + return Err(format!( + "scenario.queries[{index}].response_format is not supported for kind = \"prom_http\"" + )); + } + } + Ok(()) +} + #[derive(Debug, Deserialize, Serialize)] pub(super) struct TableConfig { pub(super) database: String, diff --git a/src/cmd/src/bin/query_perf_fixture/mod.rs b/src/cmd/src/bin/query_perf_fixture/mod.rs index 17f639bb54e..09e40b4c0dd 100644 --- a/src/cmd/src/bin/query_perf_fixture/mod.rs +++ b/src/cmd/src/bin/query_perf_fixture/mod.rs @@ -106,6 +106,7 @@ pub async fn run() { fn run_plan(args: PlanArgs) -> Result<(), Box> { let case_text = fs::read_to_string(&args.case)?; let mut case: CaseFile = toml::from_str(&case_text)?; + validate_queries(case.scenario.queries())?; if let Scenario::PromRemoteWriteThenQuery(s) = &mut case.scenario && let Some(storage) = &mut s.remote_write.storage { diff --git a/src/cmd/src/bin/query_regression_runner/measure.rs b/src/cmd/src/bin/query_regression_runner/measure.rs index fd80813f40e..6edb7bc9018 100644 --- a/src/cmd/src/bin/query_regression_runner/measure.rs +++ b/src/cmd/src/bin/query_regression_runner/measure.rs @@ -19,9 +19,13 @@ use std::{fs, io}; use reqwest::Client; use serde_json::{Map, Value, json}; -use crate::query_regression_runner::model::{Measurement, Query, QueryResult, Scenario, Table}; +use crate::query_regression_runner::model::{ + Measurement, Query, QueryResult, ResponseFormat, 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_format, sql_ident, +}; use crate::query_regression_runner::{MeasureArgs, Result}; pub(super) async fn run_measure(args: MeasureArgs) -> Result<()> { @@ -109,6 +113,8 @@ 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`). +/// SQL queries forward their configured `response_format`; SQL text is +/// unchanged. 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( @@ -122,7 +128,14 @@ async fn post_query(client: &Client, port: u16, query: &Query, db: &str) -> Valu ) .await } else { - http_post_sql(client, port, &query.query, db).await + http_post_sql_with_format( + client, + port, + &query.query, + db, + query.response_format.unwrap_or(ResponseFormat::Json), + ) + .await } } @@ -137,6 +150,7 @@ async fn run_target( queries.push(Query { name: Some("count_all".to_string()), kind: Some("sql".to_string()), + response_format: None, query: format!("SELECT count(*) FROM {}", sql_ident(&tables[0].name)), start: None, end: None, @@ -435,6 +449,7 @@ mod tests { let query = Query { name: Some("q".to_string()), kind: None, + response_format: None, query: "SELECT 1".to_string(), start: None, end: None, @@ -474,4 +489,118 @@ mod tests { ); assert_eq!(results[1]["reason"], "unsupported threshold"); } + + const MULTI_STATEMENT_SQL: &str = "INSERT INTO opt02_fulltext_flush (ts, msg) SELECT greptime_timestamp, concat('disk pressure on ', instance, ', value ', cast(greptime_value AS STRING)) FROM opt02_fulltext_source; ADMIN FLUSH_TABLE('opt02_fulltext_flush')"; + + fn batch_query(response_format: Option) -> Query { + Query { + name: Some("insert_flush".to_string()), + kind: Some("sql".to_string()), + response_format, + query: MULTI_STATEMENT_SQL.to_string(), + start: None, + end: None, + step: None, + warmup: 0, + iterations: 1, + thresholds: Map::new(), + } + } + + fn form_fields(request: &str) -> std::collections::HashMap { + let body = request.split_once("\r\n\r\n").unwrap().1; + reqwest::Url::parse(&format!("http://localhost/?{body}")) + .unwrap() + .query_pairs() + .map(|(key, value)| (key.into_owned(), value.into_owned())) + .collect() + } + + /// Minimal in-test HTTP server: one canned response body per request, the + /// captured requests returned for wire assertions. No new HTTP framework. + async fn spawn_sql_server( + responses: Vec, + ) -> (u16, tokio::task::JoinHandle>) { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let server = tokio::spawn(async move { + let mut requests = Vec::new(); + for body in responses { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = Vec::new(); + let mut chunk = [0; 1024]; + loop { + let read = stream.read(&mut chunk).await.unwrap(); + assert_ne!(read, 0, "client closed before sending its body"); + request.extend_from_slice(&chunk[..read]); + let Some(end) = request.windows(4).position(|w| w == b"\r\n\r\n") else { + continue; + }; + let headers = String::from_utf8_lossy(&request[..end]).to_ascii_lowercase(); + let length: usize = headers + .lines() + .find_map(|line| line.strip_prefix("content-length: ")) + .unwrap() + .parse() + .unwrap(); + if request.len() >= end + 4 + length { + break; + } + } + requests.push(String::from_utf8(request).unwrap()); + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\ + Connection: close\r\nContent-Length: {}\r\n\r\n{body}", + body.len() + ); + stream.write_all(response.as_bytes()).await.unwrap(); + } + requests + }); + (port, server) + } + + #[tokio::test] + async fn post_query_forwards_one_unchanged_batch_with_the_configured_format() { + let (port, server) = spawn_sql_server(vec![ + json!({"output": [{"affectedrows": 32768}, {"affectedrows": 0}], "execution_time_ms": 12}) + .to_string(), + json!({"data": [], "affected_rows": 0, "execution_time_ms": 3}).to_string(), + ]) + .await; + let client = Client::new(); + + let native = post_query( + &client, + port, + &batch_query(Some(ResponseFormat::GreptimedbV1)), + "public", + ) + .await; + assert!(native["ok"].as_bool().unwrap(), "{native}"); + // The unchanged native envelope, both statement outputs, is kept. + assert_eq!(native["response"]["output"].as_array().unwrap().len(), 2); + assert_eq!( + native["response"]["output"][0]["affectedrows"], + json!(32768) + ); + assert_eq!(native["response"]["execution_time_ms"], json!(12)); + + // No response_format keeps the historical flat default. + let flat = post_query(&client, port, &batch_query(None), "public").await; + assert!(flat["ok"].as_bool().unwrap(), "{flat}"); + + let requests = server.await.unwrap(); + assert_eq!(requests.len(), 2); + for (index, expected_format) in ["greptimedb_v1", "json"].into_iter().enumerate() { + let fields = form_fields(&requests[index]); + // The measured SQL text reaches the wire byte-for-byte, unchanged. + assert_eq!(fields["sql"], MULTI_STATEMENT_SQL); + assert_eq!(fields["format"], expected_format); + assert_eq!(fields["db"], "public"); + } + } } diff --git a/src/cmd/src/bin/query_regression_runner/model.rs b/src/cmd/src/bin/query_regression_runner/model.rs index 5fafdb5635b..c5a1d429d36 100644 --- a/src/cmd/src/bin/query_regression_runner/model.rs +++ b/src/cmd/src/bin/query_regression_runner/model.rs @@ -206,12 +206,38 @@ pub(super) struct ReadBenchConfig { pub(super) max_files: Option, } +/// Requested `/v1/sql` response format for one measured query. `Json` is the +/// historical flat body used by setup/visibility/discovery requests; +/// `GreptimedbV1` is the native `{"output": [...]}` envelope, the only body +/// that can serialize a batch with more than one statement output. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Deserialize, Serialize)] +pub(super) enum ResponseFormat { + #[serde(rename = "json")] + Json, + #[serde(rename = "greptimedb_v1")] + GreptimedbV1, +} + +impl ResponseFormat { + pub(super) const fn as_str(self) -> &'static str { + match self { + ResponseFormat::Json => "json", + ResponseFormat::GreptimedbV1 => "greptimedb_v1", + } + } +} + #[derive(Clone, Debug, Deserialize, Serialize)] pub(super) struct Query { #[serde(default)] pub(super) name: Option, #[serde(default)] pub(super) kind: Option, + /// Optional `/v1/sql` response format for SQL queries. Absent keeps the + /// historical flat `json` body. Rejected for `kind = "prom_http"` queries + /// (see `plan::validate_query_formats`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(super) response_format: Option, pub(super) query: String, /// Prometheus HTTP range query parameters (`kind = "prom_http"` only). #[serde(default)] @@ -250,3 +276,42 @@ pub(super) struct Measurement { pub(super) latency_ms_p95: Option, pub(super) status: String, } + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + #[test] + fn response_format_defaults_absent_and_rejects_unsupported_values() { + let query: Query = serde_json::from_value(json!({"query": "SELECT 1"})).unwrap(); + assert_eq!(query.response_format, None); + // Absent stays absent so historical report shapes are unchanged. + assert!( + serde_json::to_value(&query) + .unwrap() + .get("response_format") + .is_none() + ); + + let sql = "INSERT INTO t (ts, msg) SELECT ts, msg FROM src; ADMIN FLUSH_TABLE('t')"; + let query: Query = + serde_json::from_value(json!({"query": sql, "response_format": "greptimedb_v1"})) + .unwrap(); + assert_eq!(query.response_format, Some(ResponseFormat::GreptimedbV1)); + let report = serde_json::to_value(&query).unwrap(); + assert_eq!(report["response_format"], json!("greptimedb_v1")); + assert_eq!(report["query"], json!(sql)); + + for value in [json!("csv"), json!("GREPTIMEDB_V1"), json!(5), json!(true)] { + assert!( + serde_json::from_value::( + json!({"query": "SELECT 1", "response_format": value}) + ) + .is_err(), + "expected rejection for {value}" + ); + } + } +} diff --git a/src/cmd/src/bin/query_regression_runner/plan.rs b/src/cmd/src/bin/query_regression_runner/plan.rs index 4fcc5135cd8..181a1e8c55a 100644 --- a/src/cmd/src/bin/query_regression_runner/plan.rs +++ b/src/cmd/src/bin/query_regression_runner/plan.rs @@ -36,16 +36,16 @@ pub(super) fn load_plan(generator: &PathBuf, case_path: &PathBuf) -> Result Result<(Vec, Vec)> { - match scenario { + let (tables, queries) = match scenario { Scenario::DirectReadableSst { tables, layout, queries, - } => Ok((validate_direct_tables(tables, layout)?, queries)), + } => (validate_direct_tables(tables, layout)?, queries), Scenario::PromRemoteWriteThenQuery { remote_write, queries, - } => Ok(( + } => ( vec![Table { database: remote_write.database, name: remote_write.metric, @@ -58,11 +58,29 @@ pub(super) fn normalize_scenario(scenario: Scenario) -> Result<(Vec
, Vec< validate_show_create_engine: false, }], queries, - )), + ), Scenario::OtlpTraceLoad { .. } => { - Err("measure requires a query scenario, not otlp_trace_load".into()) + return Err("measure requires a query scenario, not otlp_trace_load".into()); + } + }; + validate_query_formats(&queries)?; + Ok((tables, queries)) +} + +/// Rejects a query-local `response_format` on `kind = "prom_http"` queries +/// before any request is sent. +pub(super) fn validate_query_formats(queries: &[Query]) -> Result<()> { + for (index, query) in queries.iter().enumerate() { + if query.response_format.is_some() && query.kind.as_deref() == Some("prom_http") { + let name = query.name.as_deref().unwrap_or(""); + return Err(format!( + "query {name} (index {index}) is kind = \"prom_http\"; \ + response_format is only supported for SQL queries" + ) + .into()); } } + Ok(()) } pub(super) fn validate_direct_tables(tables: Vec
, layout: Layout) -> Result> { @@ -120,3 +138,42 @@ pub(super) fn normalized_otlp_load( } } } + +#[cfg(test)] +mod tests { + use serde_json::Map; + + use super::*; + use crate::query_regression_runner::model::ResponseFormat; + + fn query(kind: &str, response_format: Option) -> Query { + Query { + name: Some("q".to_string()), + kind: Some(kind.to_string()), + response_format, + query: "SELECT 1".to_string(), + start: None, + end: None, + step: None, + warmup: 0, + iterations: 1, + thresholds: Map::new(), + } + } + + #[test] + fn response_format_is_rejected_on_prom_http_only() { + let error = + validate_query_formats(&[query("prom_http", Some(ResponseFormat::GreptimedbV1))]) + .unwrap_err(); + assert!(error.to_string().contains("prom_http"), "{error}"); + + for (kind, format) in [ + ("prom_http", None), + ("sql", None), + ("sql", Some(ResponseFormat::GreptimedbV1)), + ] { + validate_query_formats(&[query(kind, format)]).unwrap(); + } + } +} diff --git a/src/cmd/src/bin/query_regression_runner/sql.rs b/src/cmd/src/bin/query_regression_runner/sql.rs index b33c0a1c22b..c3898863697 100644 --- a/src/cmd/src/bin/query_regression_runner/sql.rs +++ b/src/cmd/src/bin/query_regression_runner/sql.rs @@ -15,9 +15,10 @@ use std::time::Instant; use reqwest::Client; -use serde_json::{Value, json}; +use serde_json::{Map, Value, json}; use crate::query_regression_runner::Result; +use crate::query_regression_runner::model::ResponseFormat; pub(super) fn sql_string(value: &str) -> String { format!("'{}'", value.replace('\'', "''")) @@ -116,11 +117,30 @@ pub(super) fn value_f64(value: Option<&Value>) -> Option { value?.as_f64().or_else(|| value?.as_str()?.parse().ok()) } +/// Posts one statement batch with the historical flat `format=json` body; its +/// parsers (setup, visibility, discovery) read the flat `response.data` shape. pub(super) async fn http_post_sql(client: &Client, port: u16, sql: &str, db: &str) -> Value { + http_post_sql_with_format(client, port, sql, db, ResponseFormat::Json).await +} + +/// Posts one statement batch with an explicit `/v1/sql` response format. The +/// SQL text is sent unchanged. +pub(super) async fn http_post_sql_with_format( + client: &Client, + port: u16, + sql: &str, + db: &str, + response_format: ResponseFormat, +) -> Value { let mut sample = post_form( client, format!("http://127.0.0.1:{port}/v1/sql"), - &[("sql", sql), ("db", db), ("format", "json")], + &[ + ("sql", sql), + ("db", db), + ("format", response_format.as_str()), + ], + response_format, ) .await; sample @@ -152,6 +172,7 @@ pub(super) async fn http_post_prom_range_query( ("end", end.unwrap_or_default()), ("step", step.unwrap_or_default()), ], + ResponseFormat::Json, ) .await; sample @@ -170,7 +191,12 @@ fn prom_range_query_url(port: u16, db: &str) -> String { url.into() } -async fn post_form(client: &Client, url: String, form: &[(&str, &str)]) -> Value { +async fn post_form( + client: &Client, + url: String, + form: &[(&str, &str)], + response_format: ResponseFormat, +) -> Value { let started = Instant::now(); let request = client.post(url).form(form); match request.send().await { @@ -178,19 +204,35 @@ async fn post_form(client: &Client, url: String, form: &[(&str, &str)]) -> Value let status = response.status().as_u16(); match response.text().await { Ok(raw) => { - let body = serde_json::from_str(&raw).unwrap_or_else(|_| json!({"raw": raw})); - let ok = status < 400 && !response_has_error(&body); + // Latency still ends after the body has been read and parsed. + let (body, malformed) = match serde_json::from_str(&raw) { + Ok(body) => (body, false), + Err(_) => (json!({"raw": raw}), true), + }; + let (ok, failure) = if status >= 400 { + (false, Some(format!("HTTP {status}"))) + } else { + match response_format { + ResponseFormat::Json => (!response_has_error(&body), None), + ResponseFormat::GreptimedbV1 => { + match native_response_error(&body, malformed) { + Some(reason) => (false, Some(reason)), + None => (true, None), + } + } + } + }; let mut sample = json!({ "ok": ok, "status": status, "latency_ms": started.elapsed().as_secs_f64() * 1000.0, "response": body, }); - if status >= 400 { + if let Some(error) = failure { sample .as_object_mut() .expect("HTTP samples are objects") - .insert("error".to_string(), Value::String(format!("HTTP {status}"))); + .insert("error".to_string(), Value::String(error)); } sample } @@ -211,6 +253,70 @@ async fn post_form(client: &Client, url: String, form: &[(&str, &str)]) -> Value } } +/// Explains why a native `greptimedb_v1` body is not fully successful, or +/// `None` when it is. The server serializes a successful batch as +/// `{output: [...], execution_time_ms}`, where every `output` entry is an +/// externally tagged `GreptimeQueryOutput` (`{"affectedrows": N}` or +/// `{"records": {...}}`), and reports failures as the root `ErrorResponse` +/// (`{code, error, execution_time_ms}`). Only the root object and the direct +/// `output` entries are inspected: record rows and their values are never +/// visited, so error-like row values cannot fail a successful response. +/// Malformed JSON, non-envelope bodies, and unknown output shapes fail. +fn native_response_error(body: &Value, malformed: bool) -> Option { + if malformed { + return Some("greptimedb_v1 response body is not valid JSON".to_string()); + } + let Some(object) = body.as_object() else { + return Some("greptimedb_v1 response body is not a JSON object".to_string()); + }; + if let Some(reason) = root_error(object) { + return Some(reason); + } + match object.get("output") { + Some(Value::Array(outputs)) => outputs.iter().enumerate().find_map(|(index, output)| { + output_error(output).map(|reason| format!("greptimedb_v1 output[{index}]: {reason}")) + }), + Some(_) => Some("greptimedb_v1 `output` is not an array".to_string()), + None => Some("greptimedb_v1 response has neither `output` nor a root error".to_string()), + } +} + +/// Status fields of the root envelope, matching the server's `ErrorResponse`. +fn root_error(envelope: &Map) -> Option { + if envelope.get("error").is_some_and(is_truthy) { + return Some("`error` is set".to_string()); + } + envelope + .get("code") + .filter(|value| !is_success_code(value)) + .map(|_| "`code` is not a success code".to_string()) +} + +/// Checks one direct `output` entry against the serialized, externally tagged +/// `GreptimeQueryOutput` shape: exactly one recognized variant whose payload +/// has that variant's type. Nothing below the payload is inspected, so row +/// values under `records` may contain arbitrary `error`/`code` fields. +fn output_error(output: &Value) -> Option { + let Some(object) = output.as_object() else { + return Some("is not a statement output object".to_string()); + }; + let Some((variant, payload)) = object.iter().next().filter(|_| object.len() == 1) else { + return Some(format!( + "must contain exactly one output variant, found {} fields", + object.len() + )); + }; + match (variant.as_str(), payload) { + ("affectedrows", payload) if payload.as_u64().is_some() => None, + ("affectedrows", _) => Some("`affectedrows` is not an unsigned integer".to_string()), + ("records", Value::Object(records)) if records.get("rows").is_some_and(Value::is_array) => { + None + } + ("records", _) => Some("`records` is not a records object with a `rows` array".to_string()), + (variant, _) => Some(format!("`{variant}` is not a recognized output variant")), + } +} + fn response_has_error(body: &Value) -> bool { let Some(body) = body.as_object() else { return false; @@ -295,4 +401,79 @@ mod tests { None ); } + + #[test] + fn native_validation_accepts_success_and_rejects_errors() { + // Successful affectedrows and records outputs; error-like row values + // below `records` are data, not failures. + for body in [ + json!({"output": [{"affectedrows": 32768}, {"affectedrows": 0}], "execution_time_ms": 12}), + json!({"output": [{"records": {"schema": {"column_schemas": []}, "rows": [], "total_rows": 0}}], "execution_time_ms": 3}), + json!({"output": [{"records": {"schema": {"column_schemas": []}, "rows": [[{"error": "row value", "code": 7}]], "total_rows": 1}}], "execution_time_ms": 3}), + json!({"output": [{"affectedrows": 1}, {"records": {"rows": [["status", "error"]]}}]}), + ] { + assert_eq!(native_response_error(&body, false), None, "{body}"); + } + + // Root errors fail even when `output` is present. + for body in [ + json!({"code": 1004, "error": "cannot output multi-statements result in json format", "output": [{"affectedrows": 1}]}), + json!({"code": 3000, "output": [{"affectedrows": 1}]}), + json!({"error": "flush failed", "output": [{"affectedrows": 1}]}), + ] { + assert!(native_response_error(&body, false).is_some(), "{body}"); + } + // A success code is tolerated. + assert_eq!( + native_response_error(&json!({"code": 0, "output": [{"affectedrows": 1}]}), false), + None + ); + + // Output entries must be exactly one recognized variant whose payload + // matches the serialized `GreptimeQueryOutput` shape. + for (body, needle) in [ + (json!({"output": [{}]}), "exactly one output variant"), + ( + json!({"output": [{"foo": 1}]}), + "`foo` is not a recognized output variant", + ), + ( + json!({"output": [{"affectedrows": 1, "records": {"rows": []}}]}), + "exactly one output variant", + ), + ( + json!({"output": [{"affectedrows": "1"}]}), + "`affectedrows` is not an unsigned integer", + ), + ( + json!({"output": [{"records": 3}]}), + "`records` is not a records object with a `rows` array", + ), + ( + json!({"output": [{"records": {}}]}), + "`records` is not a records object with a `rows` array", + ), + ( + json!({"output": ["boom"]}), + "is not a statement output object", + ), + ] { + let reason = native_response_error(&body, false).unwrap(); + assert!( + reason.starts_with("greptimedb_v1 output[0]: ") && reason.contains(needle), + "{reason}" + ); + } + + // Malformed and non-envelope 200 bodies fail. + for (body, malformed) in [ + (json!({"raw": "not json"}), true), + (json!({"raw": "not json"}), false), + (json!({"foo": 1}), false), + (json!("boom"), false), + (json!({"output": "boom"}), false), + ] { + assert!(native_response_error(&body, malformed).is_some(), "{body}"); + } + } } diff --git a/tests/perf/README.md b/tests/perf/README.md index 0361f598a3b..8dbcd6cc857 100644 --- a/tests/perf/README.md +++ b/tests/perf/README.md @@ -137,6 +137,24 @@ defaults are owned by Rust. The outer CI driver calls to select the Rust runner lifecycle. The fixture helper exposes `direct-sst`, `prom-remote-write`, and `inspect-footer` subcommands. +### Query response format + +A `[[scenario.queries]]` entry may set `response_format = "json"` (the default) +or `"greptimedb_v1"` for SQL queries. The default keeps the historical flat +`format=json` parsing (`response.data`, `response.affected_rows`), as do all +setup, visibility, and discovery requests. `greptimedb_v1` posts +`format=greptimedb_v1`; only the root object and the direct `output` statement +entries are validated (record rows and schemas are never inspected), the raw +envelope is kept in the report, and a 200 body that is not a valid native +envelope fails the sample. Opt in only when the flat formats cannot express the +response: `json` rejects a response with more than one statement output (HTTP 400 +code 1004), so a measured semicolon-separated statement batch must use +`greptimedb_v1`. The SQL text is forwarded unchanged; `response_format` is +rejected on `kind = "prom_http"` and invalid values are rejected before any +request. An insert+flush measurement that includes index creation also requires +the mito default synchronous index build (`index.build_mode = "sync"`); under +`"async"` the flush returns before the index build finishes. + Remote-write cases that need to validate storage output can add `[scenario.remote_write.storage]`. When present, the scenario body becomes: remote-write → flush → visibility check → query measurement → stop and await @@ -565,3 +583,10 @@ 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`. + +The `opt02_fulltext_flush` case measures the end-to-end insert+flush cost of a +fixed remote-write source metric into append-mode mito tables with and without a +fulltext bloom index on a STRING field. Its two measured statement batches opt +into `response_format = "greptimedb_v1"` and expect 393216 rows per table +(12 inserts of 32768 rows each), one `fulltext_bloom` index row for the indexed +table, and a `matches_term` partial filter pushdown. diff --git a/tests/perf/query_cases/opt02_fulltext_flush/case.toml b/tests/perf/query_cases/opt02_fulltext_flush/case.toml new file mode 100644 index 00000000000..e742905e38b --- /dev/null +++ b/tests/perf/query_cases/opt02_fulltext_flush/case.toml @@ -0,0 +1,149 @@ +# OPT02 fulltext index flush workload. +# +# Each measured HTTP round trip is one statement batch: `INSERT INTO
+# SELECT ... FROM ` followed by `ADMIN FLUSH_TABLE('
')` in +# the same body. The sample is end-to-end batch time (planning, distributed scan, +# insert, flush, fulltext index build), not per-row ingestion or query latency, +# and the identical batch goes to both targets. +# +# `response_format = "greptimedb_v1"` is required: the flat `json` format rejects +# a response with more than one statement output (HTTP 400, code 1004). The SQL +# text is unchanged; the single-statement diagnostics stay on flat `json`. +# +# `ADMIN FLUSH_TABLE` covers fulltext index creation only under the mito default +# synchronous index build (`index.build_mode = "sync"` in +# `src/mito2/src/config.rs`; the runner starts datanodes without a config file). +# No per-table or `ADMIN` override exists, and under `"async"` the flush returns +# before the index build finishes, so the batch would no longer include index +# creation. +# +# Expected results to inspect in the report (not automatic assertions): +# * every `insert_flush_*` sample answers 200 with one direct `output` entry +# per statement: the insert reports `affectedrows` 32768 and the flush a +# single-row `records` entry named `ADMIN FLUSH_TABLE(...)`. Including the +# validation post for `queries[0]`, each table receives 12 inserts +# (1 + 2 + 9 and 3 + 9), so both count post-checks report 393216 rows: every +# inserted row contains the term `disk`. +# * `post_check_index_metadata` reports exactly one `fulltext_bloom` row +# (`FULLTEXT_INDEX_msg`, index_type `FULLTEXT`) for `opt02_fulltext_flush` +# and none for the control table; `post_check_explain_matches_term` shows the +# `matches_term` filter with partial filter pushdown, and the verbose analyze +# runs first so it carries the cold fulltext index apply metrics. +# +# The indexed column is `msg` because `message` is a reserved DDL keyword; +# `msg STRING FULLTEXT INDEX` matches +# `tests/cases/standalone/common/create/create_with_fulltext.sql`, and the bloom +# backend option matches `tests-integration/tests/http.rs`. +[case] +name = "opt02_fulltext_flush" +description = "OPT02: end-to-end insert+flush of a fixed remote-write metric into append-mode mito tables with and without a fulltext bloom index on a STRING FIELD (multi-statement batches measured with response_format = greptimedb_v1)" + +[scenario] +kind = "prom_remote_write_then_query" + +[scenario.remote_write] +database = "public" +metric = "opt02_fulltext_source" +physical_table = "opt02_fulltext_source_physical" +# 256 series x 128 samples = 32768 source rows. +series_count = 256 +samples_per_series = 128 +sample_chunk_size = 64 +flush_every_sample_chunks = 1 +# 2024-01-01T00:00:00Z; 128 samples per series at a 60s step. +start_unix_millis = 1704067200000 +step_millis = 60000 +chunk_series_count = 128 +timeout_seconds = 180 +visibility_timeout_seconds = 120 + +# Identical on both targets; the high TWCS trigger keeps compaction from +# rewriting the flushed files during the run. +base_setup_sql = [ + "CREATE TABLE IF NOT EXISTS opt02_fulltext_source_physical (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ('physical_metric_table'='', 'compaction.twcs.trigger_file_num'='100')", + "CREATE TABLE IF NOT EXISTS opt02_fulltext_flush (ts TIMESTAMP TIME INDEX, msg STRING FULLTEXT INDEX WITH(backend='bloom')) ENGINE=mito WITH (append_mode='true', 'compaction.type'='twcs', 'compaction.twcs.trigger_file_num'='100')", + "CREATE TABLE IF NOT EXISTS opt02_fulltext_flush_control (ts TIMESTAMP TIME INDEX, msg STRING) ENGINE=mito WITH (append_mode='true', 'compaction.type'='twcs', 'compaction.twcs.trigger_file_num'='100')", +] + +candidate_setup_sql = [ + "CREATE TABLE IF NOT EXISTS opt02_fulltext_source_physical (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ('physical_metric_table'='', 'compaction.twcs.trigger_file_num'='100')", + "CREATE TABLE IF NOT EXISTS opt02_fulltext_flush (ts TIMESTAMP TIME INDEX, msg STRING FULLTEXT INDEX WITH(backend='bloom')) ENGINE=mito WITH (append_mode='true', 'compaction.type'='twcs', 'compaction.twcs.trigger_file_num'='100')", + "CREATE TABLE IF NOT EXISTS opt02_fulltext_flush_control (ts TIMESTAMP TIME INDEX, msg STRING) ENGINE=mito WITH (append_mode='true', 'compaction.type'='twcs', 'compaction.twcs.trigger_file_num'='100')", +] + +[scenario.remote_write.value] +pattern = "linear" +base = 0 +step = 0.125 + +[scenario.remote_write.prom_store] +pending_rows_flush_interval = "1s" +max_batch_rows = 100000 + +# Measured workload: multi-statement batches (see the file header for +# `response_format`). +[[scenario.queries]] +name = "insert_flush_fulltext" +kind = "sql" +response_format = "greptimedb_v1" +query = "INSERT INTO opt02_fulltext_flush (ts, msg) SELECT greptime_timestamp, concat('disk pressure on ', instance, ', value ', cast(greptime_value AS STRING)) FROM opt02_fulltext_source; ADMIN FLUSH_TABLE('opt02_fulltext_flush')" +warmup = 2 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 30 + +[[scenario.queries]] +name = "insert_flush_control" +kind = "sql" +response_format = "greptimedb_v1" +query = "INSERT INTO opt02_fulltext_flush_control (ts, msg) SELECT greptime_timestamp, concat('disk pressure on ', instance, ', value ', cast(greptime_value AS STRING)) FROM opt02_fulltext_source; ADMIN FLUSH_TABLE('opt02_fulltext_flush_control')" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 30 + +# Diagnostics: single statements on the default flat `json` format, ordered so +# the verbose analyze is the first fulltext scan of the case. +[[scenario.queries]] +name = "post_check_explain_matches_term" +kind = "sql" +query = "EXPLAIN SELECT ts, msg FROM opt02_fulltext_flush WHERE matches_term(msg, 'disk')" +warmup = 0 +iterations = 1 + +[[scenario.queries]] +name = "post_check_explain_analyze_verbose" +kind = "sql" +query = "EXPLAIN ANALYZE VERBOSE SELECT count(*) FROM opt02_fulltext_flush WHERE matches_term(msg, 'disk')" +warmup = 0 +iterations = 1 + +[[scenario.queries]] +name = "post_check_fulltext_rows" +kind = "sql" +query = "SELECT count(*) AS fulltext_rows FROM opt02_fulltext_flush" +warmup = 0 +iterations = 1 + +[[scenario.queries]] +name = "post_check_control_rows" +kind = "sql" +query = "SELECT count(*) AS control_rows FROM opt02_fulltext_flush_control" +warmup = 0 +iterations = 1 + +[[scenario.queries]] +name = "post_check_fulltext_matches" +kind = "sql" +query = "SELECT count(*) AS fulltext_matches FROM opt02_fulltext_flush WHERE matches_term(msg, 'disk')" +warmup = 0 +iterations = 1 + +[[scenario.queries]] +name = "post_check_index_metadata" +kind = "sql" +query = "SELECT table_name, index_name, column_name, index_type, greptime_index_type FROM information_schema.statistics WHERE table_schema = 'public' AND table_name IN ('opt02_fulltext_flush', 'opt02_fulltext_flush_control') ORDER BY table_name, index_name, seq_in_index" +warmup = 0 +iterations = 1