mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-12 23:12:28 +00:00
test(perf): measure fulltext insert and flush batches with native SQL response
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
@@ -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,
|
||||
|
||||
@@ -106,6 +106,7 @@ pub async fn run() {
|
||||
fn run_plan(args: PlanArgs) -> Result<(), Box<dyn std::error::Error>> {
|
||||
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
|
||||
{
|
||||
|
||||
@@ -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<ResponseFormat>) -> 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<String, String> {
|
||||
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<String>,
|
||||
) -> (u16, tokio::task::JoinHandle<Vec<String>>) {
|
||||
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");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -206,12 +206,38 @@ pub(super) struct ReadBenchConfig {
|
||||
pub(super) max_files: Option<usize>,
|
||||
}
|
||||
|
||||
/// 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<String>,
|
||||
#[serde(default)]
|
||||
pub(super) kind: Option<String>,
|
||||
/// 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<ResponseFormat>,
|
||||
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<f64>,
|
||||
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::<Query>(
|
||||
json!({"query": "SELECT 1", "response_format": value})
|
||||
)
|
||||
.is_err(),
|
||||
"expected rejection for {value}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -36,16 +36,16 @@ pub(super) fn load_plan(generator: &PathBuf, case_path: &PathBuf) -> Result<Valu
|
||||
}
|
||||
|
||||
pub(super) fn normalize_scenario(scenario: Scenario) -> Result<(Vec<Table>, Vec<Query>)> {
|
||||
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<Table>, 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("<unnamed>");
|
||||
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<Table>, layout: Layout) -> Result<Vec<Table>> {
|
||||
@@ -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<ResponseFormat>) -> 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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<f64> {
|
||||
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<String> {
|
||||
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<String, Value>) -> Option<String> {
|
||||
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<String> {
|
||||
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": "<html>not json</html>"}), true),
|
||||
(json!({"raw": "<html>not json</html>"}), false),
|
||||
(json!({"foo": 1}), false),
|
||||
(json!("boom"), false),
|
||||
(json!({"output": "boom"}), false),
|
||||
] {
|
||||
assert!(native_response_error(&body, malformed).is_some(), "{body}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
# OPT02 fulltext index flush workload.
|
||||
#
|
||||
# Each measured HTTP round trip is one statement batch: `INSERT INTO <table>
|
||||
# SELECT ... FROM <source metric>` followed by `ADMIN FLUSH_TABLE('<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
|
||||
Reference in New Issue
Block a user