ci: backport Rust query regression and ephemeral ECS runners

Backport #8651, #8937, #8986 and #9005. Preserve the five v1.2 routine cases and map tooling checks into the release CI workflow.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-09-08 20:52:51 +08:00
parent ad2b75a0c6
commit c8ace87cf0
42 changed files with 7492 additions and 2881 deletions
+7
View File
@@ -14,6 +14,11 @@ name = "query_perf_fixture"
path = "src/bin/query_perf_fixture.rs"
required-features = ["dev-tools"]
[[bin]]
name = "query_regression_runner"
path = "src/bin/query_regression_runner.rs"
required-features = ["dev-tools"]
[features]
default = [
"servers/pprof",
@@ -68,6 +73,7 @@ common-wal.workspace = true
datafusion.workspace = true
datafusion-common.workspace = true
datafusion-physical-plan.workspace = true
datafusion_object_store.workspace = true
datanode.workspace = true
datatypes.workspace = true
either = "1.15"
@@ -87,6 +93,7 @@ mito-codec.workspace = true
mito2.workspace = true
moka = { workspace = true, features = ["future"] }
object-store.workspace = true
object_store_opendal.workspace = true
parquet = { workspace = true, features = ["object_store"] }
plugins.workspace = true
prometheus.workspace = true
@@ -13,17 +13,39 @@
// limitations under the License.
use std::collections::BTreeSet;
use std::fs::{self, File};
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use clap::Args as ClapArgs;
use parquet::file::reader::{FileReader, SerializedFileReader};
use serde::Serialize;
use datafusion_object_store::path::Path as StorePath;
use datafusion_object_store::{ObjectMeta, ObjectStore};
use futures::{StreamExt, TryStreamExt};
use object_store::config::ObjectStoreConfig;
use object_store::factory::new_raw_object_store;
use object_store::services::Fs;
use parquet::arrow::async_reader::ParquetObjectReader;
use parquet::file::metadata::ParquetMetaDataReader;
use serde::{Deserialize, Serialize};
/// Same shape as `query_regression_runner::model::DestinationConfig`. The two
/// binaries are separate crate roots, so the serde-compatible struct is defined
/// here instead of being shared; it reuses the `object-store` crate's
/// `ObjectStoreConfig` deserialization.
#[derive(Debug, Deserialize)]
struct DestinationConfig {
data_home: String,
object_store: ObjectStoreConfig,
}
#[derive(Debug, ClapArgs)]
pub(super) struct InspectFooterArgs {
/// Local data home (convenience shortcut for a File destination).
#[arg(long)]
root: PathBuf,
root: Option<PathBuf>,
/// TOML file: data_home = "..." and object_store = { type = "File" | "S3" | ... }.
#[arg(long)]
destination: Option<PathBuf>,
#[arg(long, default_value = "greptime_value")]
column: String,
#[arg(long)]
@@ -68,64 +90,140 @@ struct FooterColumnChunkReport {
num_values: i64,
}
pub(super) fn run_inspect_footer(
/// A parquet data file discovered by listing the store, with the metadata needed
/// to read its footer without any extra stat/head call.
#[derive(Debug)]
struct ListedFile {
location: StorePath,
size: u64,
path: String,
relative_path: String,
}
pub(super) async fn run_inspect_footer(
args: InspectFooterArgs,
) -> Result<(), Box<dyn std::error::Error>> {
let files = collect_parquet_files(&args.root, args.include_metadata_files)?;
let reports = files
.iter()
.map(|p| inspect_file(&args.root, p, &args.column))
.collect::<Result<Vec<_>, _>>()?;
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let (destination, root) = match (args.destination, args.root) {
(Some(_), Some(_)) => {
return Err("--destination and --root are mutually exclusive; pass exactly one".into());
}
(None, None) => return Err("one of --destination or --root is required".into()),
(Some(path), None) => {
let destination: DestinationConfig = toml::from_str(&fs::read_to_string(path)?)?;
(Some(destination), None)
}
(None, root) => (None, root),
};
let store = build_store(destination.as_ref(), root.as_deref()).await?;
let root_display = root
.as_deref()
.map(|root| root.display().to_string())
.unwrap_or_else(|| {
destination
.as_ref()
.expect("destination is set when root is not")
.data_home
.clone()
});
let files =
collect_parquet_files(store.as_ref(), root.as_deref(), args.include_metadata_files).await?;
let mut reports = futures::stream::iter(files.iter())
.map(|file| inspect_file(Arc::clone(&store), file, &args.column))
.buffer_unordered(8)
.try_collect::<Vec<_>>()
.await?;
reports.sort_by(|a, b| a.relative_path.cmp(&b.relative_path));
println!(
"{}",
serde_json::to_string_pretty(&FooterReport {
root: args.root.display().to_string(),
root: root_display,
summary: summarize_footer(&args.column, &reports),
files: reports
})?
);
Ok(())
}
fn collect_parquet_files(
root: &Path,
/// Builds an arrow `object_store::ObjectStore` for the given destination. The
/// File backend (and the `--root` shortcut) uses the opendal `Fs` builder
/// directly to avoid the create_dir_all / clean_temp_dir side effects of
/// `object_store::factory::new_fs_object_store`; every other backend goes
/// through `object_store::factory::new_raw_object_store`.
async fn build_store(
destination: Option<&DestinationConfig>,
root: Option<&Path>,
) -> Result<Arc<dyn ObjectStore>, Box<dyn std::error::Error + Send + Sync>> {
let operator = match destination {
Some(destination) => match &destination.object_store {
ObjectStoreConfig::File(_) => {
object_store::ObjectStore::new(Fs::default().root(&destination.data_home))?.finish()
}
_ => new_raw_object_store(&destination.object_store, &destination.data_home).await?,
},
None => {
let root = root.expect("root must be set when destination is None");
object_store::ObjectStore::new(Fs::default().root(&root.to_string_lossy()))?.finish()
}
};
Ok(Arc::new(object_store_opendal::OpendalStore::new(operator)))
}
async fn collect_parquet_files(
store: &dyn ObjectStore,
root: Option<&Path>,
include_metadata_files: bool,
) -> Result<Vec<PathBuf>, Box<dyn std::error::Error>> {
) -> Result<Vec<ListedFile>, Box<dyn std::error::Error + Send + Sync>> {
let metas = store.list(None).try_collect::<Vec<ObjectMeta>>().await?;
let mut files = Vec::new();
collect_parquet_files_inner(root, root, include_metadata_files, &mut files)?;
files.sort();
for meta in metas {
if !is_parquet_data_file(&meta) {
continue;
}
let is_metadata = meta
.location
.parts()
.any(|part| part.as_ref() == "metadata");
if !include_metadata_files && is_metadata {
continue;
}
let relative_path = meta.location.to_string();
let path = match root {
Some(root) => root.join(&relative_path).display().to_string(),
None => relative_path.clone(),
};
files.push(ListedFile {
location: meta.location,
size: meta.size,
path,
relative_path,
});
}
files.sort_by(|a, b| a.relative_path.cmp(&b.relative_path));
Ok(files)
}
fn collect_parquet_files_inner(
root: &Path,
path: &Path,
include_metadata_files: bool,
files: &mut Vec<PathBuf>,
) -> Result<(), Box<dyn std::error::Error>> {
if path.is_dir() {
for entry in fs::read_dir(path)? {
collect_parquet_files_inner(root, &entry?.path(), include_metadata_files, files)?;
}
} else if path.extension().is_some_and(|ext| ext == "parquet")
&& (include_metadata_files
|| !path
.strip_prefix(root)
.unwrap_or(path)
.components()
.any(|c| c.as_os_str() == "metadata"))
{
files.push(path.to_path_buf());
}
Ok(())
/// Keeps only real parquet data files: `.parquet` keys with a non-zero size.
/// Zero-size keys (and keys ending in `/`) are directory markers emitted by
/// some backends and are skipped even when they happen to end in `.parquet`,
/// since an empty parquet file has no readable footer anyway.
fn is_parquet_data_file(meta: &ObjectMeta) -> bool {
meta.location.extension() == Some("parquet")
&& meta.size > 0
&& !meta.location.as_ref().ends_with('/')
}
fn inspect_file(
root: &Path,
path: &Path,
async fn inspect_file(
store: Arc<dyn ObjectStore>,
file: &ListedFile,
column: &str,
) -> Result<FooterFileReport, Box<dyn std::error::Error>> {
let file_size = fs::metadata(path)?.len();
let reader = SerializedFileReader::new(File::open(path)?)?;
let metadata = reader.metadata().file_metadata();
let row_groups = reader.metadata().row_groups();
) -> Result<FooterFileReport, Box<dyn std::error::Error + Send + Sync>> {
let mut reader =
ParquetObjectReader::new(store, file.location.clone()).with_file_size(file.size);
let metadata = ParquetMetaDataReader::new()
.load_and_finish(&mut reader, file.size)
.await?;
let file_metadata = metadata.file_metadata();
let row_groups = metadata.row_groups();
let mut columns = Vec::new();
for (row_group_index, rg) in row_groups.iter().enumerate() {
for (column_index, chunk) in rg.columns().iter().enumerate() {
@@ -145,14 +243,10 @@ fn inspect_file(
}
}
Ok(FooterFileReport {
path: path.display().to_string(),
relative_path: path
.strip_prefix(root)
.unwrap_or(path)
.display()
.to_string(),
file_size,
num_rows: metadata.num_rows(),
path: file.path.clone(),
relative_path: file.relative_path.clone(),
file_size: file.size,
num_rows: file_metadata.num_rows(),
num_row_groups: row_groups.len(),
columns,
})
@@ -179,3 +273,199 @@ fn summarize_footer(column: &str, files: &[FooterFileReport]) -> FooterSummary {
s.unique_encodings = encodings.into_iter().collect();
s
}
#[cfg(test)]
mod tests {
use std::fs::{self, File};
use parquet::column::writer::ColumnWriter;
use parquet::file::properties::WriterProperties;
use parquet::file::writer::SerializedFileWriter;
use parquet::schema::parser::parse_message_type;
use tempfile::TempDir;
use super::*;
/// Writes a minimal valid parquet file with a single `value` Int64 column
/// and `rows` values, so the footer can be read back through the object
/// store path.
fn write_parquet_file(path: &Path, rows: i64) {
let schema = Arc::new(
parse_message_type("message schema { REQUIRED INT64 value; }").expect("valid schema"),
);
let props = Arc::new(WriterProperties::builder().build());
let file = File::create(path).expect("create parquet file");
let mut writer = SerializedFileWriter::new(file, schema, props).expect("init writer");
let mut row_group = writer.next_row_group().expect("next row group");
let mut column = row_group
.next_column()
.expect("next column")
.expect("has column");
match column.untyped() {
ColumnWriter::Int64ColumnWriter(typed) => {
typed
.write_batch(&(1..=rows).collect::<Vec<_>>(), None, None)
.expect("write batch");
}
_ => panic!("unexpected column writer"),
}
column.close().expect("close column");
row_group.close().expect("close row group");
writer.close().expect("close writer");
}
fn fixture_tree(dir: &Path) {
// Nested data tree with a `metadata` subdirectory, a non-parquet file
// and a zero-size fake file.
let table = dir.join("data/table");
fs::create_dir_all(&table).expect("create table dir");
fs::create_dir_all(table.join("metadata")).expect("create metadata dir");
write_parquet_file(&table.join("0001.parquet"), 3);
write_parquet_file(&table.join("0002.parquet"), 5);
write_parquet_file(&table.join("metadata/0001.parquet"), 7);
fs::write(table.join("notes.txt"), b"not parquet").expect("write notes");
File::create(table.join("empty.parquet")).expect("create empty file");
}
/// Runs the full list -> filter -> footer-read path against a File backend
/// rooted at `dir` and returns the collected file reports.
async fn inspect_fixture(dir: &Path, include_metadata_files: bool) -> Vec<FooterFileReport> {
let root = Some(dir.to_path_buf());
let store = build_store(None, root.as_deref())
.await
.expect("build store");
let files = collect_parquet_files(store.as_ref(), root.as_deref(), include_metadata_files)
.await
.expect("collect files");
let mut reports = futures::stream::iter(files.iter())
.map(|file| inspect_file(Arc::clone(&store), file, "value"))
.buffer_unordered(8)
.try_collect::<Vec<_>>()
.await
.expect("inspect files");
reports.sort_by(|a, b| a.relative_path.cmp(&b.relative_path));
reports
}
#[tokio::test]
async fn lists_and_inspects_footer_files_without_metadata_by_default() {
let dir = TempDir::new().expect("tempdir");
fixture_tree(dir.path());
let reports = inspect_fixture(dir.path(), false).await;
let relative_paths = reports
.iter()
.map(|report| report.relative_path.as_str())
.collect::<Vec<_>>();
// metadata/0001.parquet excluded, notes.txt and empty.parquet excluded.
assert_eq!(
relative_paths,
vec!["data/table/0001.parquet", "data/table/0002.parquet"]
);
assert!(reports.iter().all(|report| report.file_size > 0));
assert!(reports.iter().all(|report| report.num_row_groups == 1));
assert!(reports.iter().all(|report| report.num_rows > 0));
assert!(
reports
.iter()
.all(|report| report.columns.len() == 1 && report.columns[0].column_path == "value")
);
// Default writer properties use dictionary encoding, so the column chunk
// reports PLAIN (dictionary page) plus RLE_DICTIONARY (data pages).
assert!(
reports
.iter()
.all(|report| report.columns[0].encodings.contains(&"PLAIN".to_string()))
);
assert!(
reports
.iter()
.all(|report| report.columns[0].compression == "UNCOMPRESSED")
);
}
#[tokio::test]
async fn includes_metadata_files_when_requested() {
let dir = TempDir::new().expect("tempdir");
fixture_tree(dir.path());
let reports = inspect_fixture(dir.path(), true).await;
let relative_paths = reports
.iter()
.map(|report| report.relative_path.as_str())
.collect::<Vec<_>>();
assert_eq!(
relative_paths,
vec![
"data/table/0001.parquet",
"data/table/0002.parquet",
"data/table/metadata/0001.parquet"
]
);
let metadata_report = reports
.iter()
.find(|report| report.relative_path == "data/table/metadata/0001.parquet")
.expect("metadata report");
assert_eq!(metadata_report.num_rows, 7);
}
#[tokio::test]
async fn accepts_a_file_destination_toml() {
let dir = TempDir::new().expect("tempdir");
fixture_tree(dir.path());
let destination = dir.path().join("destination.toml");
fs::write(
&destination,
format!(
"data_home = \"{}\"\n[object_store]\ntype = \"File\"\n",
dir.path().display()
),
)
.expect("write destination toml");
let destination: DestinationConfig =
toml::from_str(&fs::read_to_string(&destination).expect("read toml"))
.expect("parse toml");
let store = build_store(Some(&destination), None)
.await
.expect("build store");
let files = collect_parquet_files(store.as_ref(), None, false)
.await
.expect("collect files");
// Destination mode has no root prefix: paths are store-relative keys.
assert_eq!(
files
.iter()
.map(|file| file.relative_path.as_str())
.collect::<Vec<_>>(),
vec!["data/table/0001.parquet", "data/table/0002.parquet"]
);
assert!(files.iter().all(|file| file.path == file.relative_path));
}
#[test]
fn requires_exactly_one_of_root_or_destination() {
use futures::FutureExt;
let both = run_inspect_footer(InspectFooterArgs {
root: Some(PathBuf::from("/tmp/root")),
destination: Some(PathBuf::from("/tmp/destination.toml")),
column: "value".to_string(),
include_metadata_files: false,
})
.now_or_never();
assert!(matches!(both, Some(Err(_))));
let neither = run_inspect_footer(InspectFooterArgs {
root: None,
destination: None,
column: "value".to_string(),
include_metadata_files: false,
})
.now_or_never();
assert!(matches!(neither, Some(Err(_))));
}
}
+3 -3
View File
@@ -95,9 +95,9 @@ pub async fn run() {
Some(Command::PromRemoteWrite(rw)) => run_prom_remote_write(rw)
.await
.expect("prom remote write failed"),
Some(Command::InspectFooter(inspect)) => {
run_inspect_footer(inspect).expect("inspect footer failed")
}
Some(Command::InspectFooter(inspect)) => run_inspect_footer(inspect)
.await
.expect("inspect footer failed"),
Some(Command::Plan(plan)) => run_plan(plan).expect("plan failed"),
None => run_direct_sst(args.legacy).await,
}
@@ -0,0 +1,23 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#![allow(clippy::print_stderr, clippy::print_stdout)]
#[path = "query_regression_runner/mod.rs"]
mod query_regression_runner;
#[tokio::main]
async fn main() {
query_regression_runner::run().await;
}
@@ -0,0 +1,501 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::fs;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::{Duration, Instant};
use reqwest::Client;
use serde::Serialize;
use serde_json::{Value, json};
use crate::query_regression_runner::model::{Scenario, Table};
use crate::query_regression_runner::plan::{load_plan, validate_direct_tables};
use crate::query_regression_runner::sql::{
extract_rows, http_post_sql, row_u64, row_value, sql_ident, sql_string, value_text,
};
use crate::query_regression_runner::{PrepareDirectArgs, Result};
#[derive(Clone, Debug, Serialize)]
struct Discovery {
table: String,
catalog: String,
schema: String,
table_id: u64,
region_id: u64,
region_seq: u64,
table_dir: String,
region_dir: String,
peer_id: u64,
peer_addr: Value,
is_leader: Value,
status: Value,
discovery_queries: Value,
}
#[derive(Debug)]
struct PreparedTarget {
creates: Vec<Value>,
discoveries: Vec<Discovery>,
}
pub(super) async fn run_prepare_direct(args: PrepareDirectArgs) -> Result<()> {
if !args.http_timeout.is_finite() || args.http_timeout < 0.0 {
return Err("--http-timeout must be a non-negative finite number".into());
}
let case_path = args.case.canonicalize()?;
let plan = load_plan(&args.fixture_generator, &case_path)?;
let scenario_value = plan
.get("scenario")
.cloned()
.ok_or("fixture plan has no scenario")?;
let scenario: Scenario = serde_json::from_value(scenario_value)?;
let tables = direct_tables(scenario)?;
let client = Client::builder()
.timeout(Duration::from_secs_f64(args.http_timeout))
.build()?;
let base = prepare_target("base", args.base_http_port, &tables, &client).await?;
let candidate = prepare_target("candidate", args.candidate_http_port, &tables, &client).await?;
if base.discoveries.len() != candidate.discoveries.len()
|| base
.discoveries
.iter()
.zip(&candidate.discoveries)
.any(|(base, candidate)| {
base.table != candidate.table
|| base.region_id != candidate.region_id
|| base.table_dir != candidate.table_dir
})
{
return Err(format!(
"base/candidate metadata mismatch: base={:?}, candidate={:?}",
base.discoveries, candidate.discoveries
)
.into());
}
let mut fixtures = Vec::with_capacity(tables.len());
for (index, (table, discovery)) in tables.iter().zip(&base.discoveries).enumerate() {
let fixture_dir = table_fixture_dir(&args.fixture_dir, &tables, table, index);
fixtures.push(generate_direct_fixture(
&args.fixture_generator,
&case_path,
&fixture_dir,
table,
discovery,
tables.len() > 1,
args.allow_large_fixture,
)?);
}
let report = json!({
"case_path": case_path,
"scenario": "direct_readable_sst",
"base": { "create_table": base.creates, "discovery": base.discoveries },
"candidate": { "create_table": candidate.creates, "discovery": candidate.discoveries },
"fixtures": fixtures,
});
let text = format!("{}\n", serde_json::to_string_pretty(&report)?);
if let Some(path) = args.output {
fs::write(path, &text)?;
}
print!("{text}");
Ok(())
}
fn direct_tables(scenario: Scenario) -> Result<Vec<Table>> {
match scenario {
Scenario::DirectReadableSst { tables, layout, .. } => {
validate_direct_tables(tables, layout)
}
Scenario::PromRemoteWriteThenQuery { .. } => {
Err("prepare-direct requires scenario kind direct_readable_sst".into())
}
Scenario::OtlpTraceLoad { .. } => {
Err("prepare-direct requires scenario kind direct_readable_sst".into())
}
}
}
async fn prepare_target(
name: &str,
port: u16,
tables: &[Table],
client: &Client,
) -> Result<PreparedTarget> {
let mut creates = Vec::with_capacity(tables.len());
let mut discoveries = Vec::with_capacity(tables.len());
for table in tables {
let sql = create_table_sql(table)?;
let result = http_post_sql(client, port, &sql, &table.database).await;
if !result["ok"].as_bool().unwrap_or(false) {
return Err(format!("CREATE TABLE {} failed for {name}: {result}", table.name).into());
}
creates.push(json!({ "table": table.name, "sql": sql, "result": result }));
discoveries.push(discover_region(client, port, table).await?);
}
Ok(PreparedTarget {
creates,
discoveries,
})
}
fn create_table_sql(table: &Table) -> Result<String> {
if table.engine != "mito" {
return Err("prepare-direct requires table engine mito".into());
}
let time_index = table
.time_index
.as_deref()
.ok_or("direct-SST table requires time_index")?;
if table.columns.is_empty() {
return Err("direct-SST table requires columns".into());
}
let columns = table
.columns
.iter()
.map(|column| format!("{} {}", sql_ident(&column.name), column.data_type))
.collect::<Vec<_>>()
.join(",\n ");
let primary_key = table
.primary_key
.iter()
.map(|column| sql_ident(column))
.collect::<Vec<_>>()
.join(", ");
let mut options = Vec::new();
if let Some(append_mode) = table.append_mode {
options.push(("append_mode", append_mode.to_string()));
}
if let Some(sst_format) = table
.sst_format
.as_deref()
.filter(|value| !value.is_empty())
{
options.push(("sst_format", sst_format.to_string()));
}
let with = (!options.is_empty()).then(|| {
format!(
"\nWITH ({})",
options
.iter()
.map(|(key, value)| format!("'{key}'='{value}'"))
.collect::<Vec<_>>()
.join(", ")
)
});
Ok(format!(
"CREATE TABLE {} (\n {columns},\n TIME INDEX ({}),\n PRIMARY KEY ({primary_key})\n) ENGINE=mito{};",
sql_ident(&table.name),
sql_ident(time_index),
with.unwrap_or_default()
))
}
async fn discover_region(client: &Client, port: u16, table: &Table) -> Result<Discovery> {
let schema = sql_string(&table.database);
let table_name = sql_string(&table.name);
let table_sql = format!(
"SELECT table_id FROM information_schema.tables WHERE table_schema = {schema} AND table_name = {table_name}"
);
let table_result = http_post_sql(client, port, &table_sql, &table.database).await;
if !table_result["ok"].as_bool().unwrap_or(false) {
return Err(format!("table_id discovery failed: {table_result}").into());
}
let table_rows = extract_rows(table_result.get("response").unwrap_or(&Value::Null));
if table_rows.len() != 1 {
return Err(format!(
"expected one information_schema.tables row, got {}: {table_result}",
table_rows.len()
)
.into());
}
let table_id = row_u64(&table_rows[0], 0, "table_id")?;
let region_sql = format!(
"SELECT region_id, peer_id, peer_addr, is_leader, status FROM information_schema.region_peers WHERE table_schema = {schema} AND table_name = {table_name}"
);
let region_result = http_post_sql(client, port, &region_sql, &table.database).await;
if !region_result["ok"].as_bool().unwrap_or(false) {
return Err(format!("region_peers discovery failed: {region_result}").into());
}
let region_rows = extract_rows(region_result.get("response").unwrap_or(&Value::Null));
if region_rows.len() != 1 {
return Err(format!(
"expected one information_schema.region_peers row, got {}: {region_result}",
region_rows.len()
)
.into());
}
let row = &region_rows[0];
let region_id = row_u64(row, 0, "region_id")?;
let peer_id = row_u64(row, 1, "peer_id")?;
let peer_addr = row_value(row, 2, "peer_addr")
.cloned()
.unwrap_or(Value::Null);
let is_leader = row_value(row, 3, "is_leader")
.cloned()
.unwrap_or(Value::Null);
let status = row_value(row, 4, "status").cloned().unwrap_or(Value::Null);
if !matches!(
value_text(&is_leader).to_ascii_lowercase().as_str(),
"yes" | "true"
) {
return Err(format!(
"expected leader region peer, got is_leader={is_leader}: {region_result}"
)
.into());
}
if !value_text(&status).eq_ignore_ascii_case("ALIVE") {
return Err(
format!("expected ALIVE region peer, got status={status}: {region_result}").into(),
);
}
if peer_id != 0 {
return Err(format!("expected region leader on datanode peer_id=0, got {peer_id}").into());
}
if region_id >> 32 != table_id {
return Err(format!(
"table_id mismatch: tables={table_id}, region_id-derived={}",
region_id >> 32
)
.into());
}
let region_seq = region_id & u32::MAX as u64;
let (table_dir, region_dir) = storage_paths(&table.database, table_id, region_seq);
Ok(Discovery {
table: table.name.clone(),
catalog: "greptime".to_string(),
schema: table.database.clone(),
table_id,
region_id,
region_seq,
table_dir: table_dir.clone(),
region_dir,
peer_id,
peer_addr,
is_leader,
status,
discovery_queries: json!({ "table": table_result, "region": region_result }),
})
}
fn storage_paths(database: &str, table_id: u64, region_seq: u64) -> (String, String) {
let table_dir = format!("data/greptime/{database}/{table_id}/");
let region_dir = format!("{table_dir}{table_id}_{region_seq:010}");
(table_dir, region_dir)
}
fn table_fixture_dir(root: &Path, tables: &[Table], table: &Table, index: usize) -> PathBuf {
if tables.len() == 1 {
root.to_path_buf()
} else {
root.join(fixture_subdir(table, index))
}
}
fn fixture_subdir(table: &Table, index: usize) -> String {
let raw = format!("{index:02}_{}_{}", table.database, table.name);
let mut safe = String::new();
let mut replaced = false;
for character in raw.chars() {
if character.is_ascii_alphanumeric() || matches!(character, '_' | '.' | '-') {
safe.push(character);
replaced = false;
} else if !replaced {
safe.push('_');
replaced = true;
}
}
let safe = safe.trim_matches(['.', '_', '-']);
if safe.is_empty() {
format!("table_{index:02}")
} else {
safe.to_string()
}
}
fn generate_direct_fixture(
generator: &Path,
case_path: &Path,
fixture_dir: &Path,
table: &Table,
discovery: &Discovery,
multi_table: bool,
allow_large_fixture: bool,
) -> Result<Value> {
if fixture_dir.exists() {
fs::remove_dir_all(fixture_dir)?;
}
fs::create_dir_all(fixture_dir)?;
let mut command = vec![
generator.to_string_lossy().to_string(),
"direct-sst".to_string(),
"--case".to_string(),
case_path.to_string_lossy().to_string(),
"--out-dir".to_string(),
fixture_dir.to_string_lossy().to_string(),
];
if multi_table {
command.extend(["--table".to_string(), table.name.clone()]);
}
command.extend([
"--region-id".to_string(),
discovery.region_id.to_string(),
"--table-dir".to_string(),
discovery.table_dir.clone(),
]);
if allow_large_fixture {
command.push("--allow-large".to_string());
}
let started = Instant::now();
let output = Command::new(generator).args(&command[1..]).output()?;
let elapsed_seconds = started.elapsed().as_secs_f64();
let stdout = String::from_utf8_lossy(&output.stdout).to_string();
let stderr = String::from_utf8_lossy(&output.stderr).to_string();
if !output.status.success() {
return Err(format!(
"fixture generator failed: command={command:?}, returncode={:?}, elapsed_seconds={elapsed_seconds:.3}, stderr={:.2000}",
output.status.code(), stderr
)
.into());
}
let summary: Value = serde_json::from_slice(&fs::read(fixture_dir.join("summary.json"))?)?;
assert_fixture_summary(&summary, table, discovery)?;
Ok(json!({
"status": "ok",
"fixture_dir": fixture_dir,
"command": command,
"returncode": output.status.code(),
"elapsed_seconds": elapsed_seconds,
"stdout": stdout,
"stderr": stderr,
"summary": summary,
}))
}
fn assert_fixture_summary(summary: &Value, table: &Table, discovery: &Discovery) -> Result<()> {
if summary.get("table").and_then(Value::as_str) != Some(&table.name) {
return Err(format!(
"fixture table mismatch: {:?} != {}",
summary.get("table"),
table.name
)
.into());
}
if summary.get("database").and_then(Value::as_str) != Some(&table.database) {
return Err(format!(
"fixture database mismatch: {:?} != {}",
summary.get("database"),
table.database
)
.into());
}
if summary.get("region_id").and_then(|value| {
value
.as_u64()
.or_else(|| value.as_str().and_then(|value| value.parse().ok()))
}) != Some(discovery.region_id)
{
return Err(format!(
"fixture region_id mismatch: {:?} != {}",
summary.get("region_id"),
discovery.region_id
)
.into());
}
if summary.get("table_dir").and_then(Value::as_str) != Some(&discovery.table_dir) {
return Err(format!(
"fixture table_dir mismatch: {:?} != {}",
summary.get("table_dir"),
discovery.table_dir
)
.into());
}
if summary
.get("region_dir")
.and_then(Value::as_str)
.map(|value| value.trim_matches('/'))
!= Some(discovery.region_dir.trim_matches('/'))
{
return Err(format!(
"fixture region_dir mismatch: {:?} != {}",
summary.get("region_dir"),
discovery.region_dir
)
.into());
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::query_regression_runner::model::Column;
#[test]
fn creates_exact_direct_sst_table_sql() {
let table = Table {
database: "public".to_string(),
name: "metric\"name".to_string(),
engine: "mito".to_string(),
columns: vec![
Column {
name: "host".to_string(),
data_type: "STRING".to_string(),
},
Column {
name: "ts".to_string(),
data_type: "TIMESTAMP(9)".to_string(),
},
],
primary_key: vec!["host".to_string()],
time_index: Some("ts".to_string()),
append_mode: Some(true),
sst_format: Some("flat".to_string()),
validate_show_create_engine: true,
};
assert_eq!(
create_table_sql(&table).unwrap(),
"CREATE TABLE \"metric\"\"name\" (\n \"host\" STRING,\n \"ts\" TIMESTAMP(9),\n TIME INDEX (\"ts\"),\n PRIMARY KEY (\"host\")\n) ENGINE=mito\nWITH ('append_mode'='true', 'sst_format'='flat');"
);
}
#[test]
fn extracts_named_and_positional_discovery_rows() {
let rows = extract_rows(&json!({"output": [{"data": [{"TABLE_ID": "7"}]}]}));
assert_eq!(rows, vec![json!({"TABLE_ID": "7"})]);
assert_eq!(row_u64(&rows[0], 0, "table_id").unwrap(), 7);
let rows = extract_rows(&json!({"data": [[30064771073_u64, 0, "addr", "YES", "ALIVE"]]}));
assert_eq!(rows.len(), 1);
assert_eq!(row_u64(&rows[0], 0, "region_id").unwrap(), 30_064_771_073);
assert_eq!(row_u64(&rows[0], 1, "peer_id").unwrap(), 0);
assert_eq!(
value_text(row_value(&rows[0], 3, "is_leader").unwrap()),
"YES"
);
}
#[test]
fn storage_paths_include_the_table_directory() {
assert_eq!(
storage_paths("public", 1024, 0),
(
"data/greptime/public/1024/".to_string(),
"data/greptime/public/1024/1024_0000000000".to_string(),
)
);
}
}
@@ -0,0 +1,225 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::fs;
use std::path::Path;
use futures::{AsyncWriteExt as _, TryStreamExt};
use object_store::ObjectStore;
use object_store::factory::new_raw_object_store;
use object_store::services::Fs;
use crate::query_regression_runner::model::{
CopyCounts, DestinationConfig, FixtureSummary, MaterializeResult,
};
use crate::query_regression_runner::{MaterializeArgs, Result};
pub(super) async fn run_materialize(args: MaterializeArgs) -> Result<()> {
let destination: DestinationConfig = toml::from_str(&fs::read_to_string(args.destination)?)?;
let result = materialize(&args.fixture_dir, destination).await?;
println!("{}", serde_json::to_string(&result)?);
Ok(())
}
async fn materialize(
fixture_dir: &Path,
destination: DestinationConfig,
) -> Result<MaterializeResult> {
let summary: FixtureSummary =
serde_json::from_slice(&fs::read(fixture_dir.join("summary.json"))?)?;
let region_dir = validate_region_dir(&summary.region_dir)?;
let object_source = fs_operator(&fixture_dir.join("object-store"))?;
let manifest_source = fs_operator(&fixture_dir.join("manifest"))?;
let destination =
new_raw_object_store(&destination.object_store, &destination.data_home).await?;
let region_prefix = format!("{region_dir}/");
destination
.delete_with(&region_prefix)
.recursive(true)
.await?;
let object_store = copy_tree(&object_source, &destination, "/", "").await?;
let manifest = copy_tree(
&manifest_source,
&destination,
"/",
&format!("{region_dir}/manifest/"),
)
.await?;
Ok(MaterializeResult {
region_dir,
object_store,
manifest,
})
}
fn fs_operator(root: &Path) -> Result<ObjectStore> {
Ok(ObjectStore::new(Fs::default().root(&root.to_string_lossy()))?.finish())
}
fn validate_region_dir(region_dir: &str) -> Result<String> {
let region_dir = region_dir.trim_end_matches('/');
if region_dir.is_empty()
|| region_dir.starts_with('/')
|| region_dir.contains('\\')
|| region_dir
.split('/')
.any(|component| component.is_empty() || matches!(component, "." | ".."))
{
return Err(
"region_dir must be a non-empty relative OpenDAL key without dot components".into(),
);
}
Ok(region_dir.to_string())
}
async fn copy_tree(
source: &ObjectStore,
destination: &ObjectStore,
source_prefix: &str,
destination_prefix: &str,
) -> Result<CopyCounts> {
let mut lister = source.lister_with(source_prefix).recursive(true).await?;
let mut counts = CopyCounts::default();
while let Some(entry) = lister.try_next().await? {
if entry.metadata().is_dir() {
continue;
}
let source_path = entry.path().to_string();
let reader = source
.reader(&source_path)
.await?
.into_futures_async_read(0..entry.metadata().content_length())
.await?;
let mut writer = destination
.writer(&format!("{destination_prefix}{source_path}"))
.await?
.into_futures_async_write();
let bytes = futures::io::copy(reader, &mut writer).await?;
writer.close().await?;
counts.files += 1;
counts.bytes += bytes;
}
Ok(counts)
}
#[cfg(test)]
mod tests {
use std::fs;
use object_store::config::ObjectStoreConfig;
use super::*;
#[test]
fn region_dir_must_be_a_relative_opendal_key() {
assert_eq!(
validate_region_dir("data/db/region/").unwrap(),
"data/db/region"
);
for region_dir in [
"",
"/region",
".",
"a/../region",
"a/./region",
"a\\region",
"a//region",
] {
assert!(validate_region_dir(region_dir).is_err(), "{region_dir}");
}
}
#[tokio::test]
async fn materializes_fixture_from_fs_to_fs() {
let fixture = tempfile::tempdir().unwrap();
let destination = tempfile::tempdir().unwrap();
let region_dir = "data/public/metrics/00000000000000000001";
let sst = fixture
.path()
.join("object-store")
.join(region_dir)
.join("00000000000000000001.parquet");
fs::create_dir_all(sst.parent().unwrap()).unwrap();
fs::write(&sst, b"sst").unwrap();
let checkpoint = fixture
.path()
.join("manifest/00000000000000000001.checkpoint");
fs::create_dir_all(checkpoint.parent().unwrap()).unwrap();
fs::write(&checkpoint, b"checkpoint").unwrap();
fs::write(fixture.path().join("manifest/_last_checkpoint"), b"last").unwrap();
fs::write(
fixture.path().join("summary.json"),
format!(r#"{{"region_dir":"{region_dir}"}}"#),
)
.unwrap();
fs::write(fixture.path().join("files.jsonl"), "metadata").unwrap();
fs::create_dir_all(destination.path().join(region_dir)).unwrap();
fs::write(destination.path().join(region_dir).join("stale"), "stale").unwrap();
fs::write(destination.path().join("unrelated"), "keep").unwrap();
let result = materialize(
fixture.path(),
DestinationConfig {
data_home: destination.path().to_string_lossy().to_string(),
object_store: ObjectStoreConfig::default(),
},
)
.await
.unwrap();
assert_eq!(result.object_store.files, 1);
assert_eq!(result.object_store.bytes, 3);
assert_eq!(result.manifest.files, 2);
assert_eq!(result.manifest.bytes, 14);
assert_eq!(
fs::read(
destination
.path()
.join(region_dir)
.join("00000000000000000001.parquet")
)
.unwrap(),
b"sst"
);
assert_eq!(
fs::read(
destination
.path()
.join(region_dir)
.join("manifest/00000000000000000001.checkpoint")
)
.unwrap(),
b"checkpoint"
);
assert_eq!(
fs::read(
destination
.path()
.join(region_dir)
.join("manifest/_last_checkpoint")
)
.unwrap(),
b"last"
);
assert!(!destination.path().join(region_dir).join("stale").exists());
assert_eq!(
fs::read(destination.path().join("unrelated")).unwrap(),
b"keep"
);
assert!(!destination.path().join("summary.json").exists());
assert!(!destination.path().join("files.jsonl").exists());
}
}
@@ -0,0 +1,423 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::collections::HashMap;
use std::time::Duration;
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::plan::{load_plan, normalize_scenario};
use crate::query_regression_runner::sql::{http_post_sql, sql_ident};
use crate::query_regression_runner::{MeasureArgs, Result};
pub(super) async fn run_measure(args: MeasureArgs) -> Result<()> {
if !args.http_timeout.is_finite() || args.http_timeout < 0.0 {
return Err("--http-timeout must be a non-negative finite number".into());
}
let case_path = args.case.canonicalize()?;
let plan = load_plan(&args.fixture_generator, &case_path)?;
let case_text = fs::read_to_string(&case_path)?;
let raw_case: toml::Value = toml::from_str(&case_text)?;
let case_metadata = raw_case
.get("case")
.cloned()
.unwrap_or(toml::Value::Table(toml::map::Map::new()));
let case_metadata = serde_json::to_value(case_metadata)?;
let scenario_value = plan
.get("scenario")
.cloned()
.ok_or("fixture plan has no scenario")?;
let scenario: Scenario = serde_json::from_value(scenario_value.clone())?;
let (tables, configured_queries) = normalize_scenario(scenario)?;
let client = Client::builder()
.timeout(Duration::from_secs_f64(args.http_timeout))
.build()?;
let base = run_target(args.base_http_port, &tables, &configured_queries, &client).await;
let candidate = run_target(
args.candidate_http_port,
&tables,
&configured_queries,
&client,
)
.await;
let thresholds = enforce_thresholds(&configured_queries, &base, &candidate)?;
let status = if base.status == "failed"
|| candidate.status == "failed"
|| thresholds
.iter()
.any(|threshold| threshold["status"] == "failed")
{
"failed"
} else {
"ok"
};
let report = json!({
"case_path": case_path,
"case": case_metadata,
"scenario": scenario_value,
"queries": configured_queries,
"query_mode": "endpoint",
"http_timeout": args.http_timeout,
"targets": [target_report("base", args.base_http_port, base), target_report("candidate", args.candidate_http_port, candidate)],
"thresholds": thresholds,
"status": status,
});
let text = format!("{}\n", serde_json::to_string_pretty(&report)?);
if let Some(path) = args.output {
fs::write(path, &text)?;
}
print!("{text}");
if status == "failed" {
std::process::exit(1);
}
Ok(())
}
fn target_report(name: &str, http_port: u16, result: QueryResult) -> Value {
let status = if result.status == "ok" {
"measured"
} else {
"failed"
};
json!({
"name": name,
"http_port": http_port,
"validation": result.validation,
"validation_errors": result.validation_errors,
"measurements": result.measurements,
"status": status,
})
}
async fn run_target(
port: u16,
tables: &[Table],
configured_queries: &[Query],
client: &Client,
) -> QueryResult {
let mut queries = configured_queries.to_vec();
if queries.is_empty() {
queries.push(Query {
name: Some("count_all".to_string()),
kind: Some("sql".to_string()),
query: format!("SELECT count(*) FROM {}", sql_ident(&tables[0].name)),
warmup: 0,
iterations: 1,
thresholds: Map::new(),
});
}
let db = &tables[0].database;
let mut validation = Vec::new();
let mut validation_errors = Vec::new();
for table in tables {
let sql = format!("SHOW CREATE TABLE {}", sql_ident(&table.name));
let sample = http_post_sql(client, port, &sql, &table.database).await;
if !sample["ok"].as_bool().unwrap_or(false) {
validation_errors.push(json!({
"sql": sql,
"error": sample.get("error"),
"response": sample.get("response"),
}));
} else {
for error in validate_show_create(&sample, table) {
validation_errors.push(json!({
"sql": sql,
"error": error,
"response": sample.get("response"),
}));
}
}
validation.push(sample);
}
let first = http_post_sql(client, port, &queries[0].query, db).await;
if !first["ok"].as_bool().unwrap_or(false) {
validation_errors.push(json!({
"sql": queries[0].query,
"error": first.get("error"),
"response": first.get("response"),
}));
}
validation.push(first);
let mut measurements = Vec::with_capacity(queries.len());
for query in &queries {
for _ in 0..query.warmup {
let warmup = http_post_sql(client, port, &query.query, db).await;
if !warmup["ok"].as_bool().unwrap_or(false) {
validation_errors.push(json!({
"sql": query.query,
"phase": "warmup",
"error": warmup.get("error"),
"response": warmup.get("response"),
}));
}
}
let mut samples = Vec::with_capacity(query.iterations);
let mut good_latencies = Vec::with_capacity(query.iterations);
for _ in 0..query.iterations {
let mut sample = http_post_sql(client, port, &query.query, db).await;
let execution_time = sample
.get("response")
.and_then(extract_execution_time)
.cloned()
.unwrap_or(Value::Null);
sample
.as_object_mut()
.expect("HTTP samples are objects")
.insert("execution_time_ms".to_string(), execution_time);
if sample["ok"].as_bool().unwrap_or(false) {
good_latencies.push(sample["latency_ms"].as_f64().unwrap_or_default());
}
samples.push(sample);
}
let median = (!good_latencies.is_empty()).then(|| median(&good_latencies));
let p95 = (!good_latencies.is_empty()).then(|| percentile(&good_latencies, 95.0));
let status = if good_latencies.len() == samples.len() {
"ok"
} else {
"failed"
};
measurements.push(Measurement {
name: query.name.clone(),
kind: query.kind.clone(),
iterations: samples.len(),
samples,
latency_ms_median: median,
latency_ms_p95: p95,
status: status.to_string(),
});
}
let failed = !validation_errors.is_empty() || measurements.iter().any(|m| m.status == "failed");
QueryResult {
validation,
validation_errors,
measurements,
status: if failed { "failed" } else { "ok" }.to_string(),
}
}
fn validate_show_create(result: &Value, table: &Table) -> Vec<&'static str> {
let text = result
.get("response")
.map(response_text)
.unwrap_or_default()
.to_lowercase();
let mut errors = Vec::new();
if !text.contains(&table.name.to_lowercase()) {
errors.push("SHOW CREATE output does not contain table name");
}
if table.validate_show_create_engine && (!text.contains("engine") || !text.contains("mito")) {
errors.push("SHOW CREATE output does not mention ENGINE=mito");
}
if table.append_mode.is_some() && !text.contains("append_mode") {
errors.push("SHOW CREATE output does not mention append_mode");
}
if table.sst_format.is_some() && !text.contains("sst_format") {
errors.push("SHOW CREATE output does not mention sst_format");
}
errors
}
fn response_text(body: &Value) -> String {
body.as_str()
.map(ToOwned::to_owned)
.unwrap_or_else(|| serde_json::to_string(body).unwrap_or_default())
}
fn extract_execution_time(body: &Value) -> Option<&Value> {
match body {
Value::Object(map) => {
for key in ["execution_time_ms", "execution_time", "elapsed"] {
if let Some(value) = map.get(key) {
return Some(value);
}
}
map.values().find_map(extract_execution_time)
}
Value::Array(values) => values.iter().find_map(extract_execution_time),
_ => None,
}
}
pub(super) fn median(values: &[f64]) -> f64 {
let mut ordered = values.to_vec();
ordered.sort_by(f64::total_cmp);
let middle = ordered.len() / 2;
if ordered.len().is_multiple_of(2) {
(ordered[middle - 1] + ordered[middle]) / 2.0
} else {
ordered[middle]
}
}
fn percentile(values: &[f64], pct: f64) -> f64 {
if values.is_empty() {
return 0.0;
}
let mut ordered = values.to_vec();
ordered.sort_by(f64::total_cmp);
let index = round_ties_even((pct / 100.0) * (ordered.len() - 1) as f64)
.clamp(0, ordered.len() as isize - 1) as usize;
ordered[index]
}
pub(super) fn round_ties_even(value: f64) -> isize {
let floor = value.floor();
let fraction = value - floor;
if fraction < 0.5 {
floor as isize
} else if fraction > 0.5 {
floor as isize + 1
} else if (floor as isize) % 2 == 0 {
floor as isize
} else {
floor as isize + 1
}
}
fn enforce_thresholds(
queries: &[Query],
base: &QueryResult,
candidate: &QueryResult,
) -> Result<Vec<Value>> {
let base_by_name: HashMap<_, _> = base
.measurements
.iter()
.map(|measurement| (measurement.name.as_deref(), measurement))
.collect();
let mut results = Vec::new();
for candidate_measurement in &candidate.measurements {
let query = queries
.iter()
.find(|query| query.name == candidate_measurement.name);
let thresholds = query.map_or_else(Map::new, |query| query.thresholds.clone());
let base_measurement = base_by_name.get(&candidate_measurement.name.as_deref());
if let Some(limit) = thresholds
.get("max_candidate_latency_regression_pct")
.filter(|value| !value.is_null())
.map(value_as_f64)
.transpose()?
{
let result = match base_measurement {
None => {
json!({"query": candidate_measurement.name, "threshold": "max_candidate_latency_regression_pct", "status": "failed", "reason": "missing base measurement"})
}
Some(base) if matches!(base.latency_ms_median, None | Some(0.0)) => {
json!({"query": candidate_measurement.name, "threshold": "max_candidate_latency_regression_pct", "status": "failed", "reason": "base median latency is missing or zero", "base_latency_ms_median": base.latency_ms_median})
}
Some(_) if candidate_measurement.latency_ms_median.is_none() => {
json!({"query": candidate_measurement.name, "threshold": "max_candidate_latency_regression_pct", "status": "failed", "reason": "missing candidate measurement"})
}
Some(base) => {
let actual = (candidate_measurement.latency_ms_median.unwrap()
- base.latency_ms_median.unwrap())
/ base.latency_ms_median.unwrap()
* 100.0;
json!({"query": candidate_measurement.name, "threshold": "max_candidate_latency_regression_pct", "status": if actual <= limit { "passed" } else { "failed" }, "actual_pct": actual, "limit_pct": limit})
}
};
results.push(result);
}
for key in thresholds
.keys()
.filter(|key| *key != "max_candidate_latency_regression_pct")
{
results.push(json!({"query": candidate_measurement.name, "threshold": key, "status": "failed", "reason": "unsupported threshold"}));
}
}
Ok(results)
}
fn value_as_f64(value: &Value) -> Result<f64> {
value.as_f64().ok_or_else(|| {
io::Error::new(io::ErrorKind::InvalidData, "threshold must be numeric").into()
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn median_and_p95_match_python() {
assert_eq!(median(&[1.0, 8.0, 3.0, 4.0]), 3.5);
assert_eq!(
percentile(
&[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0],
95.0
),
10.0
);
assert_eq!(percentile(&[], 95.0), 0.0);
}
#[test]
fn finds_nested_execution_time_in_priority_order() {
let body = json!({"output": [{"elapsed": 3}], "execution_time": 2});
assert_eq!(extract_execution_time(&body), Some(&json!(2)));
assert_eq!(
extract_execution_time(&json!({"output": [{"elapsed": 3}]})),
Some(&json!(3))
);
}
#[test]
fn threshold_rejects_unknown_keys_and_zero_base() {
let query = Query {
name: Some("q".to_string()),
kind: None,
query: "SELECT 1".to_string(),
warmup: 0,
iterations: 1,
thresholds: Map::from_iter([
("max_candidate_latency_regression_pct".to_string(), json!(0)),
("other".to_string(), json!(1)),
]),
};
let measurement = |median| Measurement {
name: Some("q".to_string()),
kind: None,
iterations: 1,
samples: vec![],
latency_ms_median: median,
latency_ms_p95: median,
status: "ok".to_string(),
};
let base = QueryResult {
validation: vec![],
validation_errors: vec![],
measurements: vec![measurement(Some(0.0))],
status: "ok".to_string(),
};
let candidate = QueryResult {
validation: vec![],
validation_errors: vec![],
measurements: vec![measurement(Some(1.0))],
status: "ok".to_string(),
};
let results = enforce_thresholds(&[query], &base, &candidate).unwrap();
assert_eq!(
results[0]["reason"],
"base median latency is missing or zero"
);
assert_eq!(results[1]["reason"], "unsupported threshold");
}
}
@@ -0,0 +1,204 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::error::Error;
use std::path::PathBuf;
use clap::{Parser, Subcommand};
mod direct;
mod materialize;
mod measure;
mod model;
mod otlp;
mod plan;
mod remote;
mod sql;
pub(super) type Result<T> = std::result::Result<T, Box<dyn Error>>;
#[derive(Debug, Parser)]
#[command(about = "Measure query endpoints or materialize a direct-SST fixture")]
struct Cli {
#[command(subcommand)]
command: RunnerCommand,
}
#[derive(Debug, Subcommand)]
enum RunnerCommand {
/// Run normalized query regression cases against two HTTP endpoints.
Measure(MeasureArgs),
/// Create direct-SST tables, discover their regions, and generate fixtures.
PrepareDirect(PrepareDirectArgs),
/// Render the frontend Prometheus store configuration for a remote-write case.
RenderRemoteConfig(RenderRemoteConfigArgs),
/// Create a database and ingest a normalized Prometheus remote-write case.
PrepareRemote(PrepareRemoteArgs),
/// Inspect stopped remote-write storage and finalize the aggregate report.
FinalizeRemote(FinalizeRemoteArgs),
/// Run one externally managed OTLP trace-load target.
RunOtlpTarget(RunOtlpTargetArgs),
/// Combine externally managed OTLP trace-load target results.
FinalizeOtlp(FinalizeOtlpArgs),
/// Copy a direct-SST fixture into an object store destination.
Materialize(MaterializeArgs),
}
#[derive(Debug, Parser)]
struct MeasureArgs {
#[arg(long, value_name = "PATH")]
case: PathBuf,
#[arg(long, value_name = "PATH")]
fixture_generator: PathBuf,
#[arg(long)]
base_http_port: u16,
#[arg(long)]
candidate_http_port: u16,
#[arg(long, value_name = "PATH")]
output: Option<PathBuf>,
#[arg(long, default_value_t = 120.0)]
http_timeout: f64,
}
#[derive(Debug, Parser)]
struct PrepareDirectArgs {
#[arg(long, value_name = "PATH")]
case: PathBuf,
#[arg(long, value_name = "PATH")]
fixture_generator: PathBuf,
#[arg(long)]
base_http_port: u16,
#[arg(long)]
candidate_http_port: u16,
#[arg(long, value_name = "PATH")]
fixture_dir: PathBuf,
#[arg(long, value_name = "PATH")]
output: Option<PathBuf>,
#[arg(long, default_value_t = 120.0)]
http_timeout: f64,
#[arg(long)]
allow_large_fixture: bool,
}
#[derive(Debug, Parser)]
struct RenderRemoteConfigArgs {
#[arg(long, value_name = "PATH")]
case: PathBuf,
#[arg(long, value_name = "PATH")]
fixture_generator: PathBuf,
#[arg(long, value_name = "PATH")]
output: PathBuf,
}
#[derive(Debug, Parser)]
struct PrepareRemoteArgs {
#[arg(long, value_name = "PATH")]
case: PathBuf,
#[arg(long, value_name = "PATH")]
fixture_generator: PathBuf,
#[arg(long)]
base_http_port: u16,
#[arg(long)]
candidate_http_port: u16,
#[arg(long, value_name = "PATH")]
output: Option<PathBuf>,
#[arg(long, default_value_t = 120.0)]
http_timeout: f64,
}
#[derive(Debug, Parser)]
struct FinalizeRemoteArgs {
#[arg(long, value_name = "PATH")]
case: PathBuf,
#[arg(long, value_name = "PATH")]
fixture_generator: PathBuf,
#[arg(long, value_name = "PATH")]
candidate_bin: PathBuf,
#[arg(long, value_name = "PATH")]
base_data_home: Option<PathBuf>,
#[arg(long, value_name = "PATH")]
candidate_data_home: Option<PathBuf>,
#[arg(long, value_name = "PATH")]
base_destination: Option<PathBuf>,
#[arg(long, value_name = "PATH")]
candidate_destination: Option<PathBuf>,
#[arg(long, value_name = "PATH")]
report: PathBuf,
}
#[derive(Debug, Parser)]
struct RunOtlpTargetArgs {
#[arg(long, value_name = "PATH")]
case: PathBuf,
#[arg(long, value_name = "PATH")]
fixture_generator: PathBuf,
#[arg(long, value_name = "PATH")]
otelgen_bin: PathBuf,
#[arg(long)]
http_port: u16,
#[arg(long)]
target_name: String,
#[arg(long, value_name = "PATH")]
work_dir: PathBuf,
#[arg(long, value_name = "PATH")]
output: PathBuf,
#[arg(long, default_value_t = 120.0)]
http_timeout: f64,
}
#[derive(Debug, Parser)]
struct FinalizeOtlpArgs {
#[arg(long, value_name = "PATH")]
case: PathBuf,
#[arg(long, value_name = "PATH")]
fixture_generator: PathBuf,
#[arg(long, value_name = "PATH")]
base_result: PathBuf,
#[arg(long, value_name = "PATH")]
candidate_result: PathBuf,
#[arg(long, value_name = "PATH")]
output: PathBuf,
}
#[derive(Debug, Parser)]
struct MaterializeArgs {
#[arg(long, value_name = "PATH")]
fixture_dir: PathBuf,
#[arg(
long,
value_name = "PATH",
help = "TOML file: data_home = \"...\" and object_store = { type = \"File\" }; object_store uses object_store::config::ObjectStoreConfig"
)]
destination: PathBuf,
}
pub(super) async fn run() {
if let Err(error) = run_inner().await {
eprintln!("query_regression_runner: {error}");
std::process::exit(1);
}
}
async fn run_inner() -> Result<()> {
match Cli::parse().command {
RunnerCommand::Measure(args) => measure::run_measure(args).await,
RunnerCommand::PrepareDirect(args) => direct::run_prepare_direct(args).await,
RunnerCommand::RenderRemoteConfig(args) => remote::run_render_remote_config(args).await,
RunnerCommand::PrepareRemote(args) => remote::run_prepare_remote(args).await,
RunnerCommand::FinalizeRemote(args) => remote::run_finalize_remote(args).await,
RunnerCommand::RunOtlpTarget(args) => otlp::run_otlp_target(args).await,
RunnerCommand::FinalizeOtlp(args) => otlp::run_finalize_otlp(args).await,
RunnerCommand::Materialize(args) => materialize::run_materialize(args).await,
}
}
@@ -0,0 +1,235 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use object_store::config::ObjectStoreConfig;
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};
#[derive(Debug, Deserialize)]
pub(super) struct DestinationConfig {
pub(super) data_home: String,
pub(super) object_store: ObjectStoreConfig,
}
#[derive(Debug, Deserialize)]
pub(super) struct FixtureSummary {
pub(super) region_dir: String,
}
#[derive(Debug, Default, Serialize)]
pub(super) struct CopyCounts {
pub(super) files: u64,
pub(super) bytes: u64,
}
#[derive(Debug, Serialize)]
pub(super) struct MaterializeResult {
pub(super) region_dir: String,
pub(super) object_store: CopyCounts,
pub(super) manifest: CopyCounts,
}
#[derive(Debug, Deserialize)]
#[serde(tag = "kind")]
pub(super) enum Scenario {
#[serde(rename = "direct_readable_sst")]
DirectReadableSst {
tables: Vec<Table>,
layout: Layout,
#[serde(default)]
queries: Vec<Query>,
},
#[serde(rename = "prom_remote_write_then_query")]
PromRemoteWriteThenQuery {
remote_write: RemoteWrite,
#[serde(default)]
queries: Vec<Query>,
},
#[serde(rename = "otlp_trace_load")]
OtlpTraceLoad { load: OtlpTraceLoad },
}
#[derive(Debug, Deserialize)]
pub(super) struct Layout {
pub(super) regions: usize,
}
#[derive(Clone, Debug, Deserialize)]
pub(super) struct Table {
pub(super) database: String,
pub(super) name: String,
#[serde(default)]
pub(super) engine: String,
#[serde(default)]
pub(super) columns: Vec<Column>,
#[serde(default)]
pub(super) primary_key: Vec<String>,
#[serde(default)]
pub(super) time_index: Option<String>,
#[serde(default)]
pub(super) append_mode: Option<bool>,
#[serde(default)]
pub(super) sst_format: Option<String>,
#[serde(default = "default_show_create_engine")]
pub(super) validate_show_create_engine: bool,
}
#[derive(Clone, Debug, Deserialize)]
pub(super) struct Column {
pub(super) name: String,
#[serde(rename = "type")]
pub(super) data_type: String,
}
const fn default_show_create_engine() -> bool {
true
}
#[derive(Clone, Debug, Deserialize)]
pub(super) struct RemoteWrite {
pub(super) database: String,
pub(super) metric: String,
pub(super) physical_table: String,
pub(super) series_count: u64,
pub(super) samples_per_series: u64,
pub(super) start_unix_millis: i64,
pub(super) step_millis: i64,
pub(super) chunk_series_count: u64,
pub(super) timeout_seconds: u64,
pub(super) sample_chunk_size: Option<u64>,
pub(super) flush_every_sample_chunks: u64,
pub(super) visibility_timeout_seconds: u64,
pub(super) prom_store: PromStore,
pub(super) value: RemoteValue,
pub(super) storage: Option<StorageConfig>,
pub(super) read_bench: Option<ReadBenchConfig>,
}
#[derive(Clone, Debug, Deserialize)]
pub(super) struct PromStore {
pub(super) pending_rows_flush_interval: String,
pub(super) max_batch_rows: u64,
pub(super) max_concurrent_flushes: u64,
pub(super) worker_channel_capacity: u64,
pub(super) max_inflight_requests: u64,
}
#[derive(Clone, Debug, Deserialize)]
pub(super) struct RemoteValue {
pub(super) pattern: String,
pub(super) base: f64,
pub(super) step: f64,
pub(super) cardinality: u64,
pub(super) seed: u64,
pub(super) run_length: u64,
pub(super) stall_every: u64,
pub(super) stall_length: u64,
pub(super) mixed_every: u64,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
pub(super) struct OtlpTraceLoad {
pub(super) load: OtlpLoad,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
pub(super) struct OtlpLoad {
pub(super) database: String,
pub(super) table: String,
pub(super) pipeline: String,
pub(super) duration_seconds: u64,
pub(super) warmup_seconds: u64,
pub(super) rate: u64,
pub(super) workers: usize,
pub(super) exporter_shards: usize,
pub(super) workload: String,
pub(super) visibility_timeout_seconds: u64,
pub(super) thresholds: OtlpThresholds,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
pub(super) struct OtlpThresholds {
pub(super) max_candidate_throughput_regression_pct: f64,
pub(super) max_candidate_mean_latency_regression_pct: f64,
pub(super) max_failure_count: u64,
}
#[derive(Clone, Debug, Deserialize)]
pub(super) struct StorageConfig {
pub(super) inspect: bool,
pub(super) column: String,
pub(super) root_suffix: Option<String>,
pub(super) include_metadata_files: bool,
pub(super) min_files: u64,
pub(super) min_files_with_column: u64,
pub(super) require_encodings: Vec<String>,
pub(super) forbid_encodings: Vec<String>,
pub(super) max_total_file_size_bytes: Option<u64>,
pub(super) max_column_compressed_size_bytes: Option<u64>,
pub(super) max_column_uncompressed_size_bytes: Option<u64>,
pub(super) max_candidate_total_file_size_regression_pct: Option<f64>,
pub(super) max_candidate_column_compressed_size_regression_pct: Option<f64>,
pub(super) max_candidate_column_uncompressed_size_regression_pct: Option<f64>,
}
#[derive(Clone, Debug, Deserialize)]
pub(super) struct ReadBenchConfig {
pub(super) enabled: bool,
pub(super) parquetbench: bool,
pub(super) scanbench: bool,
pub(super) iterations: u64,
pub(super) projection: Vec<String>,
pub(super) parquet_reader: String,
pub(super) scan_scanner: String,
pub(super) parallelism: u64,
pub(super) max_files: Option<usize>,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
pub(super) struct Query {
#[serde(default)]
pub(super) name: Option<String>,
#[serde(default)]
pub(super) kind: Option<String>,
pub(super) query: String,
#[serde(default)]
pub(super) warmup: usize,
#[serde(default = "one")]
pub(super) iterations: usize,
#[serde(default)]
pub(super) thresholds: Map<String, Value>,
}
const fn one() -> usize {
1
}
#[derive(Debug, Serialize)]
pub(super) struct QueryResult {
pub(super) validation: Vec<Value>,
pub(super) validation_errors: Vec<Value>,
pub(super) measurements: Vec<Measurement>,
pub(super) status: String,
}
#[derive(Debug, Serialize)]
pub(super) struct Measurement {
pub(super) name: Option<String>,
pub(super) kind: Option<String>,
pub(super) iterations: usize,
pub(super) samples: Vec<Value>,
pub(super) latency_ms_median: Option<f64>,
pub(super) latency_ms_p95: Option<f64>,
pub(super) status: String,
}
@@ -0,0 +1,618 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::collections::HashMap;
use std::fs;
use std::path::Path;
use std::process::{Command, Stdio};
use std::time::{Duration, Instant};
use regex::Regex;
use reqwest::Client;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use crate::query_regression_runner::measure::round_ties_even;
use crate::query_regression_runner::model::{OtlpLoad, OtlpThresholds};
use crate::query_regression_runner::plan::normalized_otlp_load;
use crate::query_regression_runner::sql::{
extract_count_value, http_post_sql, sql_ident, sql_string, value_f64, value_u64,
};
use crate::query_regression_runner::{FinalizeOtlpArgs, Result, RunOtlpTargetArgs};
const OTLP_ROWS: &str = "greptime_frontend_otlp_traces_rows";
const OTLP_FAILURES: &str = "greptime_frontend_otlp_traces_failure_count";
const OTLP_ELAPSED_SUM: &str = "greptime_servers_http_otlp_traces_elapsed_sum";
const OTLP_ELAPSED_COUNT: &str = "greptime_servers_http_otlp_traces_elapsed_count";
#[derive(Clone, Debug, Deserialize, Serialize)]
struct OtlpSnapshot {
captured_monotonic_seconds: f64,
values: HashMap<String, f64>,
}
pub(super) async fn run_otlp_target(args: RunOtlpTargetArgs) -> Result<()> {
if !args.http_timeout.is_finite() || args.http_timeout < 0.0 {
return Err("--http-timeout must be a non-negative finite number".into());
}
let (_, load) = normalized_otlp_load(&args.fixture_generator, &args.case)?;
let client = Client::builder()
.timeout(Duration::from_secs_f64(args.http_timeout))
.build()?;
let result = match run_otlp_target_inner(&args, &load, &client).await {
Ok(result) => result,
Err(error) => {
json!({ "name": args.target_name, "status": "failed", "error": error.to_string() })
}
};
let text = format!("{}\n", serde_json::to_string_pretty(&result)?);
fs::write(&args.output, &text)?;
print!("{text}");
if result["status"] == "failed" {
std::process::exit(1);
}
Ok(())
}
async fn run_otlp_target_inner(
args: &RunOtlpTargetArgs,
load: &OtlpLoad,
client: &Client,
) -> Result<Value> {
let create_database = http_post_sql(
client,
args.http_port,
&format!(
"CREATE DATABASE IF NOT EXISTS {}",
sql_ident(&load.database)
),
"public",
)
.await;
if !create_database["ok"].as_bool().unwrap_or(false) {
return Ok(
json!({ "name": args.target_name, "create_database": create_database, "status": "failed" }),
);
}
let otelgen = run_otelgen_load(
&args.otelgen_bin,
args.http_port,
&args.work_dir,
load,
client,
)
.await?;
let metrics = summarize_otlp_metrics(&otelgen)?;
let flush = http_post_sql(
client,
args.http_port,
&format!("ADMIN FLUSH_TABLE({})", sql_string(&load.table)),
&load.database,
)
.await;
let visibility = poll_otlp_visibility(
client,
args.http_port,
&load.table,
&load.database,
metrics["accepted_spans"].as_u64().unwrap_or_default(),
load.visibility_timeout_seconds,
)
.await?;
let checks_ok = otelgen["status"] == "ok"
&& metrics["missing_metrics"]
.as_array()
.is_some_and(Vec::is_empty)
&& metrics["accepted_spans"].as_u64().unwrap_or_default() > 0
&& metrics["http_requests"].as_u64().unwrap_or_default() > 0
&& flush["ok"].as_bool().unwrap_or(false)
&& visibility["ok"].as_bool().unwrap_or(false)
&& visibility["row_count_ok"].as_bool().unwrap_or(false);
Ok(json!({
"name": args.target_name,
"otelgen": otelgen,
"metrics": metrics,
"create_database": create_database,
"flush": flush,
"visibility": visibility,
"status": if checks_ok { "measured" } else { "failed" },
}))
}
fn otelgen_command(otelgen_bin: &Path, http_port: u16, load: &OtlpLoad) -> Vec<String> {
vec![
otelgen_bin.to_string_lossy().to_string(),
"--protocol".to_string(),
"http".to_string(),
"--otel-exporter-otlp-endpoint".to_string(),
format!("127.0.0.1:{http_port}"),
"--otel-exporter-otlp-url-path".to_string(),
"/v1/otlp/v1/traces".to_string(),
"--header".to_string(),
format!("x-greptime-pipeline-name={}", load.pipeline),
"--header".to_string(),
format!("x-greptime-db-name={}", load.database),
"--header".to_string(),
format!("x-greptime-trace-table-name={}", load.table),
"--log-level".to_string(),
"error".to_string(),
"--insecure".to_string(),
"--duration".to_string(),
load.duration_seconds.to_string(),
"--rate".to_string(),
load.rate.to_string(),
"traces".to_string(),
"multi".to_string(),
"--workers".to_string(),
load.workers.to_string(),
"--scenarios".to_string(),
load.workload.clone(),
"--exporter-shards".to_string(),
load.exporter_shards.to_string(),
]
}
async fn run_otelgen_load(
otelgen_bin: &Path,
http_port: u16,
work_dir: &Path,
load: &OtlpLoad,
client: &Client,
) -> Result<Value> {
let command = otelgen_command(otelgen_bin, http_port, load);
let clock = Instant::now();
let initial = fetch_otlp_metrics(client, http_port, &clock).await?;
let log_dir = work_dir.join("otelgen");
fs::create_dir_all(&log_dir)?;
let stdout_path = log_dir.join("stdout.log");
let stderr_path = log_dir.join("stderr.log");
let mut child = Command::new(otelgen_bin)
.args(&command[1..])
.stdout(Stdio::from(fs::File::create(&stdout_path)?))
.stderr(Stdio::from(fs::File::create(&stderr_path)?))
.spawn()?;
let started = Instant::now();
if load.warmup_seconds > 0 {
let _ = wait_for_child(&mut child, Duration::from_secs(load.warmup_seconds)).await?;
}
let warmed = fetch_otlp_metrics(client, http_port, &clock).await?;
let mut timed_out = false;
if child.try_wait()?.is_none() {
let remaining = load
.duration_seconds
.saturating_sub(load.warmup_seconds)
.saturating_add(60)
.max(60);
if !wait_for_child(&mut child, Duration::from_secs(remaining)).await? {
timed_out = true;
child.kill()?;
let _ = child.wait()?;
}
}
let final_snapshot = fetch_otlp_metrics(client, http_port, &clock).await?;
let returncode = match child.try_wait()? {
Some(status) => status.code(),
None => {
child.kill()?;
child.wait()?.code()
}
};
let elapsed_seconds = started.elapsed().as_secs_f64();
Ok(json!({
"status": if returncode == Some(0) && !timed_out && elapsed_seconds + 1.0 >= load.duration_seconds as f64 { "ok" } else { "failed" },
"cmd": command,
"returncode": returncode,
"timed_out": timed_out,
"elapsed_seconds": elapsed_seconds,
"stdout_path": stdout_path,
"stderr_path": stderr_path,
"snapshots": { "initial": initial, "warmup": warmed, "final": final_snapshot },
}))
}
async fn wait_for_child(child: &mut std::process::Child, timeout: Duration) -> Result<bool> {
let deadline = Instant::now() + timeout;
loop {
if child.try_wait()?.is_some() {
return Ok(true);
}
if Instant::now() >= deadline {
return Ok(false);
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
}
async fn fetch_otlp_metrics(
client: &Client,
http_port: u16,
clock: &Instant,
) -> Result<OtlpSnapshot> {
let text = client
.get(format!("http://127.0.0.1:{http_port}/metrics"))
.send()
.await?
.error_for_status()?
.text()
.await?;
Ok(OtlpSnapshot {
captured_monotonic_seconds: clock.elapsed().as_secs_f64(),
values: parse_prometheus_metrics(&text)?,
})
}
fn parse_prometheus_metrics(text: &str) -> Result<HashMap<String, f64>> {
let sample = Regex::new(r"^([a-zA-Z_:][a-zA-Z0-9_:]*)(?:\{[^}]*\})?\s+(\S+)")?;
let mut values = HashMap::new();
for line in text.lines() {
let Some(captures) = sample.captures(line.trim()) else {
continue;
};
let (Some(name), Some(value)) = (captures.get(1), captures.get(2)) else {
continue;
};
let name = name.as_str();
if !matches!(
name,
OTLP_ROWS | OTLP_FAILURES | OTLP_ELAPSED_SUM | OTLP_ELAPSED_COUNT
) {
continue;
}
let value: f64 = value.as_str().parse()?;
if !value.is_finite() {
return Err(format!("non-finite Prometheus sample for {name}").into());
}
*values.entry(name.to_string()).or_default() += value;
}
Ok(values)
}
fn metric_delta(after: &OtlpSnapshot, before: &OtlpSnapshot, name: &str) -> Result<f64> {
let delta = after.values.get(name).copied().unwrap_or_default()
- before.values.get(name).copied().unwrap_or_default();
if delta < 0.0 {
return Err(format!("metric {name} decreased by {}", -delta).into());
}
Ok(delta)
}
fn summarize_otlp_metrics(run: &Value) -> Result<Value> {
let snapshots = run
.get("snapshots")
.ok_or("otelgen result has no snapshots")?;
let initial: OtlpSnapshot = serde_json::from_value(snapshots["initial"].clone())?;
let warmed: OtlpSnapshot = serde_json::from_value(snapshots["warmup"].clone())?;
let final_snapshot: OtlpSnapshot = serde_json::from_value(snapshots["final"].clone())?;
let missing_metrics = [OTLP_ROWS, OTLP_ELAPSED_SUM, OTLP_ELAPSED_COUNT]
.into_iter()
.filter(|name| !final_snapshot.values.contains_key(*name))
.collect::<Vec<_>>();
let accepted_spans =
round_ties_even(metric_delta(&final_snapshot, &initial, OTLP_ROWS)?) as u64;
let measurement_accepted_spans =
round_ties_even(metric_delta(&final_snapshot, &warmed, OTLP_ROWS)?) as u64;
let http_requests =
round_ties_even(metric_delta(&final_snapshot, &warmed, OTLP_ELAPSED_COUNT)?) as u64;
let latency_seconds = metric_delta(&final_snapshot, &warmed, OTLP_ELAPSED_SUM)?;
let measurement_seconds =
final_snapshot.captured_monotonic_seconds - warmed.captured_monotonic_seconds;
Ok(json!({
"accepted_spans": accepted_spans,
"measurement_accepted_spans": measurement_accepted_spans,
"accepted_spans_per_second": if measurement_seconds > 0.0 { Some(measurement_accepted_spans as f64 / measurement_seconds) } else { None },
"http_requests": http_requests,
"mean_http_latency_ms": if http_requests > 0 { Some(latency_seconds / http_requests as f64 * 1000.0) } else { None },
"failure_count": round_ties_even(metric_delta(&final_snapshot, &initial, OTLP_FAILURES)?) as u64,
"measurement_seconds": measurement_seconds,
"missing_metrics": missing_metrics,
}))
}
async fn poll_otlp_visibility(
client: &Client,
port: u16,
table_name: &str,
database: &str,
expected_rows: u64,
timeout_seconds: u64,
) -> Result<Value> {
let sql = format!("SELECT count(*) FROM {}", sql_ident(table_name));
let deadline = Instant::now() + Duration::from_secs(timeout_seconds);
let mut attempts = 0;
loop {
attempts += 1;
let mut result = http_post_sql(client, port, &sql, database).await;
let observed_rows = extract_count_value(&result);
let row_count_ok =
result["ok"].as_bool().unwrap_or(false) && observed_rows == Some(expected_rows);
result
.as_object_mut()
.ok_or("count result must be an object")?
.extend([
("expected_rows".to_string(), json!(expected_rows)),
("observed_rows".to_string(), json!(observed_rows)),
("attempts".to_string(), json!(attempts)),
("row_count_ok".to_string(), json!(row_count_ok)),
]);
if row_count_ok {
return Ok(result);
}
if Instant::now() >= deadline {
result["ok"] = json!(false);
result["row_count_ok"] = json!(false);
result["error"] = json!(format!(
"expected {expected_rows} rows but observed {observed_rows:?} after {attempts} attempts"
));
return Ok(result);
}
tokio::time::sleep(Duration::from_millis(500)).await;
}
}
pub(super) async fn run_finalize_otlp(args: FinalizeOtlpArgs) -> Result<()> {
let (case_path, load) = normalized_otlp_load(&args.fixture_generator, &args.case)?;
let raw_case: toml::Value = toml::from_str(&fs::read_to_string(&case_path)?)?;
let case_metadata = serde_json::to_value(
raw_case
.get("case")
.cloned()
.unwrap_or(toml::Value::Table(toml::map::Map::new())),
)?;
let base: Value = serde_json::from_slice(&fs::read(&args.base_result)?)?;
let candidate: Value = serde_json::from_slice(&fs::read(&args.candidate_result)?)?;
let thresholds = enforce_otlp_thresholds(
&load.thresholds,
base.get("metrics").unwrap_or(&Value::Null),
candidate.get("metrics").unwrap_or(&Value::Null),
);
let failed = thresholds
.iter()
.any(|threshold| threshold["status"] == "failed")
|| [base.get("status"), candidate.get("status")]
.into_iter()
.any(|status| status.and_then(Value::as_str) == Some("failed"));
let report = json!({
"case_path": case_path,
"case": case_metadata,
"scenario": { "kind": "otlp_trace_load", "load": load },
"targets": [base, candidate],
"thresholds": thresholds,
"status": if failed { "failed" } else { "ok" },
});
let text = format!("{}\n", serde_json::to_string_pretty(&report)?);
fs::write(&args.output, &text)?;
print!("{text}");
if failed {
std::process::exit(1);
}
Ok(())
}
fn enforce_otlp_thresholds(
thresholds: &OtlpThresholds,
base: &Value,
candidate: &Value,
) -> Vec<Value> {
let mut results = Vec::new();
for (target, metrics) in [("base", base), ("candidate", candidate)] {
let failures = value_u64(metrics.get("failure_count"));
results.push(json!({
"target": target,
"threshold": "max_failure_count",
"status": if failures.is_some_and(|failures| failures <= thresholds.max_failure_count) { "passed" } else { "failed" },
"actual": failures,
"limit": thresholds.max_failure_count,
}));
}
for (name, base_value, candidate_value, limit, inverse) in [
(
"max_candidate_throughput_regression_pct",
value_f64(base.get("accepted_spans_per_second")),
value_f64(candidate.get("accepted_spans_per_second")),
thresholds.max_candidate_throughput_regression_pct,
true,
),
(
"max_candidate_mean_latency_regression_pct",
value_f64(base.get("mean_http_latency_ms")),
value_f64(candidate.get("mean_http_latency_ms")),
thresholds.max_candidate_mean_latency_regression_pct,
false,
),
] {
let (Some(base_value), Some(candidate_value)) =
(base_value.filter(|value| *value != 0.0), candidate_value)
else {
results.push(json!({
"threshold": name,
"status": "failed",
"reason": if inverse { "missing or zero throughput" } else { "missing or zero mean latency" },
"base": base_value,
"candidate": candidate_value,
}));
continue;
};
let actual = if inverse {
(base_value - candidate_value) / base_value * 100.0
} else {
(candidate_value - base_value) / base_value * 100.0
};
results.push(json!({
"threshold": name,
"status": if actual <= limit { "passed" } else { "failed" },
"actual_pct": actual,
"limit_pct": limit,
"base": base_value,
"candidate": candidate_value,
}));
}
results
}
#[cfg(test)]
mod tests {
use super::*;
fn otlp_load_for_test() -> OtlpLoad {
serde_json::from_value(json!({
"database": "public",
"table": "opentelemetry_traces",
"pipeline": "greptime_trace_v1",
"duration_seconds": 120,
"warmup_seconds": 60,
"rate": 50000,
"workers": 4,
"exporter_shards": 4,
"workload": "microservices",
"visibility_timeout_seconds": 30,
"thresholds": {
"max_candidate_throughput_regression_pct": 20.0,
"max_candidate_mean_latency_regression_pct": 20.0,
"max_failure_count": 0
}
}))
.unwrap()
}
#[test]
fn parses_labeled_otlp_metrics_and_rejects_non_finite_samples() {
let metrics = parse_prometheus_metrics(
"greptime_frontend_otlp_traces_rows 10\n\
greptime_frontend_otlp_traces_failure_count{kind=\"a\"} 1\n\
greptime_frontend_otlp_traces_failure_count{kind=\"b\"} 2\n",
)
.unwrap();
assert_eq!(metrics[OTLP_ROWS], 10.0);
assert_eq!(metrics[OTLP_FAILURES], 3.0);
assert!(parse_prometheus_metrics("greptime_frontend_otlp_traces_rows NaN").is_err());
let before = OtlpSnapshot {
captured_monotonic_seconds: 0.0,
values: HashMap::from([(OTLP_ROWS.to_string(), 2.0)]),
};
let after = OtlpSnapshot {
captured_monotonic_seconds: 1.0,
values: HashMap::from([(OTLP_ROWS.to_string(), 1.0)]),
};
assert!(metric_delta(&after, &before, OTLP_ROWS).is_err());
}
#[test]
fn summarizes_otlp_deltas_and_builds_exact_command() {
let snapshot = |captured: f64, values: Vec<(String, f64)>| OtlpSnapshot {
captured_monotonic_seconds: captured,
values: values.into_iter().collect(),
};
let initial = snapshot(0.0, vec![(OTLP_ROWS.to_string(), 10.0)]);
let warmed = snapshot(
5.0,
vec![
(OTLP_ROWS.to_string(), 110.0),
(OTLP_ELAPSED_SUM.to_string(), 1.0),
(OTLP_ELAPSED_COUNT.to_string(), 10.0),
],
);
let final_snapshot = snapshot(
15.0,
vec![
(OTLP_ROWS.to_string(), 310.0),
(OTLP_FAILURES.to_string(), 3.0),
(OTLP_ELAPSED_SUM.to_string(), 3.0),
(OTLP_ELAPSED_COUNT.to_string(), 30.0),
],
);
let metrics = summarize_otlp_metrics(&json!({
"snapshots": { "initial": initial, "warmup": warmed, "final": final_snapshot }
}))
.unwrap();
assert_eq!(metrics["accepted_spans"], 300);
assert_eq!(metrics["measurement_accepted_spans"], 200);
assert_eq!(metrics["accepted_spans_per_second"], 20.0);
assert_eq!(metrics["http_requests"], 20);
assert_eq!(metrics["mean_http_latency_ms"], 100.0);
assert_eq!(metrics["failure_count"], 3);
let missing = summarize_otlp_metrics(&json!({
"snapshots": {
"initial": snapshot(0.0, vec![(OTLP_ROWS.to_string(), 0.0)]),
"warmup": snapshot(1.0, vec![(OTLP_ROWS.to_string(), 1.0)]),
"final": snapshot(2.0, vec![(OTLP_ROWS.to_string(), 2.0)])
}
}))
.unwrap();
assert_eq!(
missing["missing_metrics"],
json!([OTLP_ELAPSED_SUM, OTLP_ELAPSED_COUNT])
);
let command = otelgen_command(Path::new("/bin/otelgen"), 4000, &otlp_load_for_test());
assert_eq!(
command,
vec![
"/bin/otelgen",
"--protocol",
"http",
"--otel-exporter-otlp-endpoint",
"127.0.0.1:4000",
"--otel-exporter-otlp-url-path",
"/v1/otlp/v1/traces",
"--header",
"x-greptime-pipeline-name=greptime_trace_v1",
"--header",
"x-greptime-db-name=public",
"--header",
"x-greptime-trace-table-name=opentelemetry_traces",
"--log-level",
"error",
"--insecure",
"--duration",
"120",
"--rate",
"50000",
"traces",
"multi",
"--workers",
"4",
"--scenarios",
"microservices",
"--exporter-shards",
"4",
]
);
}
#[test]
fn otlp_thresholds_handle_zero_metrics_and_pass_fail_cases() {
let load = otlp_load_for_test();
let base = json!({ "failure_count": 0, "accepted_spans_per_second": 20.0, "mean_http_latency_ms": 100.0 });
let candidate = json!({ "failure_count": 0, "accepted_spans_per_second": 18.0, "mean_http_latency_ms": 110.0 });
let results = enforce_otlp_thresholds(&load.thresholds, &base, &candidate);
assert!(results.iter().all(|result| result["status"] == "passed"));
let failed = enforce_otlp_thresholds(
&load.thresholds,
&json!({ "failure_count": 0, "accepted_spans_per_second": 0.0, "mean_http_latency_ms": 100.0 }),
&json!({ "failure_count": 3, "accepted_spans_per_second": 20.0, "mean_http_latency_ms": 130.0 }),
);
assert!(
failed
.iter()
.any(|result| result["threshold"] == "max_failure_count"
&& result["target"] == "candidate"
&& result["status"] == "failed")
);
assert!(failed.iter().any(|result| result["threshold"]
== "max_candidate_throughput_regression_pct"
&& result["reason"] == "missing or zero throughput"));
assert!(failed.iter().any(|result| result["threshold"]
== "max_candidate_mean_latency_regression_pct"
&& result["status"] == "failed"));
}
}
@@ -0,0 +1,122 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::process::Command;
use serde_json::Value;
use crate::query_regression_runner::Result;
use crate::query_regression_runner::model::{
Layout, OtlpLoad, Query, RemoteWrite, Scenario, Table,
};
pub(super) fn load_plan(generator: &PathBuf, case_path: &PathBuf) -> Result<Value> {
let output = Command::new(generator)
.args(["plan", "--case"])
.arg(case_path)
.output()?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
return Err(format!("query_perf_fixture plan failed: {:.2000}", stderr).into());
}
Ok(serde_json::from_slice(&output.stdout)?)
}
pub(super) fn normalize_scenario(scenario: Scenario) -> Result<(Vec<Table>, Vec<Query>)> {
match scenario {
Scenario::DirectReadableSst {
tables,
layout,
queries,
} => Ok((validate_direct_tables(tables, layout)?, queries)),
Scenario::PromRemoteWriteThenQuery {
remote_write,
queries,
} => Ok((
vec![Table {
database: remote_write.database,
name: remote_write.metric,
engine: "metric".to_string(),
columns: vec![],
primary_key: vec![],
time_index: None,
append_mode: None,
sst_format: None,
validate_show_create_engine: false,
}],
queries,
)),
Scenario::OtlpTraceLoad { .. } => {
Err("measure requires a query scenario, not otlp_trace_load".into())
}
}
}
pub(super) fn validate_direct_tables(tables: Vec<Table>, layout: Layout) -> Result<Vec<Table>> {
if tables.is_empty() || layout.regions != 1 {
return Err("runner supports one or more tables and exactly one region per table".into());
}
let mut pairs = HashSet::new();
let mut names = HashSet::new();
for table in &tables {
if !pairs.insert((&table.database, &table.name)) {
return Err("duplicate (database, name) table entries are not supported".into());
}
if !names.insert(&table.name) {
return Err("duplicate table names are not supported".into());
}
}
Ok(tables)
}
pub(super) fn normalized_remote_write(
generator: &PathBuf,
case: &Path,
) -> Result<(PathBuf, RemoteWrite)> {
let case_path = case.canonicalize()?;
let plan = load_plan(generator, &case_path)?;
let scenario = plan
.get("scenario")
.cloned()
.ok_or("fixture plan has no scenario")?;
match serde_json::from_value(scenario)? {
Scenario::PromRemoteWriteThenQuery { remote_write, .. } => Ok((case_path, remote_write)),
Scenario::DirectReadableSst { .. } => {
Err("remote command requires scenario kind prom_remote_write_then_query".into())
}
Scenario::OtlpTraceLoad { .. } => {
Err("remote command requires scenario kind prom_remote_write_then_query".into())
}
}
}
pub(super) fn normalized_otlp_load(
generator: &PathBuf,
case: &Path,
) -> Result<(PathBuf, OtlpLoad)> {
let case_path = case.canonicalize()?;
let plan = load_plan(generator, &case_path)?;
let scenario = plan
.get("scenario")
.cloned()
.ok_or("fixture plan has no scenario")?;
match serde_json::from_value(scenario)? {
Scenario::OtlpTraceLoad { load } => Ok((case_path, load.load)),
Scenario::DirectReadableSst { .. } | Scenario::PromRemoteWriteThenQuery { .. } => {
Err("OTLP command requires scenario kind otlp_trace_load".into())
}
}
}
@@ -0,0 +1,505 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::fs;
use std::path::Path;
use std::process::Command;
use std::time::{Duration, Instant};
use reqwest::Client;
use serde_json::{Value, json};
use crate::query_regression_runner::model::{PromStore, RemoteWrite};
use crate::query_regression_runner::plan::normalized_remote_write;
use crate::query_regression_runner::sql::{
extract_count_value, http_post_sql, sql_ident, sql_string, value_f64, value_u64,
};
use crate::query_regression_runner::{PrepareRemoteArgs, RenderRemoteConfigArgs, Result};
pub(super) async fn run_render_remote_config(args: RenderRemoteConfigArgs) -> Result<()> {
let (_, remote) = normalized_remote_write(&args.fixture_generator, &args.case)?;
fs::write(args.output, frontend_prom_config(&remote.prom_store)?)?;
Ok(())
}
fn frontend_prom_config(prom: &PromStore) -> Result<String> {
Ok(format!(
"[prom_store]\nenable = true\nwith_metric_engine = true\npending_rows_flush_interval = {}\nmax_batch_rows = {}\nmax_concurrent_flushes = {}\nworker_channel_capacity = {}\nmax_inflight_requests = {}\n",
serde_json::to_string(&prom.pending_rows_flush_interval)?,
prom.max_batch_rows,
prom.max_concurrent_flushes,
prom.worker_channel_capacity,
prom.max_inflight_requests,
))
}
pub(super) async fn run_prepare_remote(args: PrepareRemoteArgs) -> Result<()> {
if !args.http_timeout.is_finite() || args.http_timeout < 0.0 {
return Err("--http-timeout must be a non-negative finite number".into());
}
let (case_path, remote) = normalized_remote_write(&args.fixture_generator, &args.case)?;
let client = Client::builder()
.timeout(Duration::from_secs_f64(args.http_timeout))
.build()?;
let base = prepare_remote_target(
"base",
args.base_http_port,
&args.fixture_generator,
&remote,
&client,
)
.await?;
let candidate = prepare_remote_target(
"candidate",
args.candidate_http_port,
&args.fixture_generator,
&remote,
&client,
)
.await?;
let report = json!({
"case_path": case_path,
"scenario": "prom_remote_write_then_query",
"base": base,
"candidate": candidate,
"status": "ok",
});
let text = format!("{}\n", serde_json::to_string_pretty(&report)?);
if let Some(path) = args.output {
fs::write(path, &text)?;
}
print!("{text}");
Ok(())
}
async fn prepare_remote_target(
name: &str,
port: u16,
generator: &Path,
remote: &RemoteWrite,
client: &Client,
) -> Result<Value> {
let create_database = http_post_sql(
client,
port,
&format!(
"CREATE DATABASE IF NOT EXISTS {}",
sql_ident(&remote.database)
),
"public",
)
.await;
if !create_database["ok"].as_bool().unwrap_or(false) {
return Err(format!(
"CREATE DATABASE {} failed for {name}: {create_database}",
remote.database
)
.into());
}
let (remote_write, flushes) = ingest_remote_write(generator, port, remote, client).await?;
let expected_rows = remote
.series_count
.checked_mul(remote.samples_per_series)
.ok_or("expected remote-write row count overflows u64")?;
let visibility = poll_expected_count(
client,
port,
&remote.metric,
&remote.database,
expected_rows,
remote.visibility_timeout_seconds,
)
.await?;
Ok(json!({
"name": name,
"create_database": create_database,
"remote_write": remote_write,
"flushes": flushes,
"visibility": visibility,
"status": "ok",
}))
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct SampleChunk {
index: u64,
offset: u64,
samples_per_series: u64,
start_unix_millis: i64,
}
async fn ingest_remote_write(
generator: &Path,
port: u16,
remote: &RemoteWrite,
client: &Client,
) -> Result<(Value, Vec<Value>)> {
let Some(chunks) = sample_chunks(remote)? else {
let remote_write = run_remote_write(generator, port, remote, None)?;
let flush = flush_remote_table(client, port, remote, "final", None).await?;
return Ok((remote_write, vec![flush]));
};
let mut results = Vec::with_capacity(chunks.len());
let mut flushes = Vec::new();
let scheduled_flushes = scheduled_flushes(&chunks, remote.flush_every_sample_chunks);
for chunk in &chunks {
let mut result = run_remote_write(generator, port, remote, Some(chunk))?;
result
.as_object_mut()
.ok_or("remote-write result must be an object")?
.extend([
("sample_offset".to_string(), json!(chunk.offset)),
(
"samples_per_series".to_string(),
json!(chunk.samples_per_series),
),
("chunk_index".to_string(), json!(chunk.index)),
]);
results.push(result);
if let Some((_, reason)) = scheduled_flushes
.iter()
.find(|(index, _)| *index == chunk.index)
{
flushes
.push(flush_remote_table(client, port, remote, reason, Some(chunk.index)).await?);
}
}
Ok((
json!({
"status": "ok",
"mode": "sample-chunked",
"sample_chunk_size": remote.sample_chunk_size,
"flush_every_sample_chunks": remote.flush_every_sample_chunks,
"chunks": results,
"aggregate": summarize_remote_chunks(&results),
}),
flushes,
))
}
fn sample_chunks(remote: &RemoteWrite) -> Result<Option<Vec<SampleChunk>>> {
let Some(chunk_samples) = remote.sample_chunk_size else {
return Ok(None);
};
if chunk_samples == 0 {
return Err("scenario.remote_write.sample_chunk_size must be positive".into());
}
if remote.flush_every_sample_chunks == 0 {
return Err("scenario.remote_write.flush_every_sample_chunks must be positive".into());
}
let mut chunks = Vec::new();
let mut offset = 0;
while offset < remote.samples_per_series {
let samples_per_series = chunk_samples.min(remote.samples_per_series - offset);
chunks.push(SampleChunk {
index: chunks.len() as u64 + 1,
offset,
samples_per_series,
start_unix_millis: remote.start_unix_millis + offset as i64 * remote.step_millis,
});
offset += samples_per_series;
}
Ok(Some(chunks))
}
fn scheduled_flushes(chunks: &[SampleChunk], flush_every: u64) -> Vec<(u64, &'static str)> {
let mut scheduled = chunks
.iter()
.filter(|chunk| chunk.index % flush_every == 0)
.map(|chunk| (chunk.index, "periodic"))
.collect::<Vec<_>>();
if let Some(chunk) = chunks.last()
&& chunk.index % flush_every != 0
{
scheduled.push((chunk.index, "final"));
}
scheduled
}
fn summarize_remote_chunks(chunks: &[Value]) -> Value {
let mut rows = 0;
let mut samples_written = 0;
let mut batches = 0;
let mut elapsed_seconds = 0.0;
for chunk in chunks {
let summary = chunk.get("summary").unwrap_or(&Value::Null);
let chunk_rows = value_u64(summary.get("rows")).unwrap_or(0);
rows += chunk_rows;
samples_written += value_u64(summary.get("samples_written")).unwrap_or(chunk_rows);
batches += value_u64(summary.get("batches")).unwrap_or(0);
elapsed_seconds += value_f64(summary.get("elapsed_seconds"))
.or_else(|| value_f64(chunk.get("elapsed_seconds")))
.unwrap_or(0.0);
}
json!({ "rows": rows, "samples_written": samples_written, "batches": batches, "elapsed_seconds": elapsed_seconds })
}
fn run_remote_write(
generator: &Path,
port: u16,
remote: &RemoteWrite,
chunk: Option<&SampleChunk>,
) -> Result<Value> {
let command = remote_write_command(generator, port, remote, chunk);
let started = Instant::now();
let output = Command::new(generator).args(&command[1..]).output()?;
let elapsed_seconds = started.elapsed().as_secs_f64();
let stdout = String::from_utf8_lossy(&output.stdout).to_string();
let stderr = String::from_utf8_lossy(&output.stderr).to_string();
if !output.status.success() {
return Err(format!(
"remote-write generator failed: command={command:?}, returncode={:?}, elapsed_seconds={elapsed_seconds:.3}, stderr={:.2000}",
output.status.code(), stderr
)
.into());
}
let mut result = json!({
"status": "ok",
"command": command,
"returncode": output.status.code(),
"elapsed_seconds": elapsed_seconds,
"stdout": stdout,
"stderr": stderr,
});
match serde_json::from_str(result["stdout"].as_str().unwrap_or_default()) {
Ok(summary) => result["summary"] = summary,
Err(_) => result["summary_parse_error"] = result["stdout"].clone(),
}
Ok(result)
}
fn remote_write_command(
generator: &Path,
port: u16,
remote: &RemoteWrite,
chunk: Option<&SampleChunk>,
) -> Vec<String> {
let (samples_per_series, start_unix_millis, sample_offset, total_samples_per_series) = chunk
.map(|chunk| {
(
chunk.samples_per_series,
chunk.start_unix_millis,
Some(chunk.offset),
Some(remote.samples_per_series),
)
})
.unwrap_or((
remote.samples_per_series,
remote.start_unix_millis,
None,
None,
));
let mut command = vec![
generator.to_string_lossy().to_string(),
"prom-remote-write".to_string(),
"--endpoint".to_string(),
format!("http://127.0.0.1:{port}/v1/prometheus/write"),
"--database".to_string(),
remote.database.clone(),
"--metric".to_string(),
remote.metric.clone(),
"--physical-table".to_string(),
remote.physical_table.clone(),
"--series-count".to_string(),
remote.series_count.to_string(),
"--samples-per-series".to_string(),
samples_per_series.to_string(),
"--start-unix-millis".to_string(),
start_unix_millis.to_string(),
"--step-millis".to_string(),
remote.step_millis.to_string(),
"--chunk-series-count".to_string(),
remote.chunk_series_count.to_string(),
"--timeout-seconds".to_string(),
remote.timeout_seconds.to_string(),
"--value-pattern".to_string(),
remote.value.pattern.clone(),
"--value-base".to_string(),
json_number(remote.value.base),
"--value-step".to_string(),
json_number(remote.value.step),
"--value-cardinality".to_string(),
remote.value.cardinality.to_string(),
"--value-seed".to_string(),
remote.value.seed.to_string(),
"--value-run-length".to_string(),
remote.value.run_length.to_string(),
"--value-stall-every".to_string(),
remote.value.stall_every.to_string(),
"--value-stall-length".to_string(),
remote.value.stall_length.to_string(),
"--value-mixed-every".to_string(),
remote.value.mixed_every.to_string(),
];
if let Some(sample_offset) = sample_offset {
command.extend([
"--value-sample-offset".to_string(),
sample_offset.to_string(),
]);
}
if let Some(total_samples_per_series) = total_samples_per_series {
command.extend([
"--value-total-samples-per-series".to_string(),
total_samples_per_series.to_string(),
]);
}
command
}
async fn flush_remote_table(
client: &Client,
port: u16,
remote: &RemoteWrite,
reason: &str,
chunk_index: Option<u64>,
) -> Result<Value> {
let mut result = http_post_sql(
client,
port,
&format!("ADMIN FLUSH_TABLE({})", sql_string(&remote.physical_table)),
&remote.database,
)
.await;
if !result["ok"].as_bool().unwrap_or(false) {
return Err(format!(
"ADMIN FLUSH_TABLE {} failed: {result}",
remote.physical_table
)
.into());
}
result
.as_object_mut()
.ok_or("flush result must be an object")?
.extend([
("physical_table".to_string(), json!(remote.physical_table)),
("reason".to_string(), json!(reason)),
("chunk_index".to_string(), json!(chunk_index)),
]);
Ok(result)
}
async fn poll_expected_count(
client: &Client,
port: u16,
table_name: &str,
database: &str,
expected_rows: u64,
visibility_timeout_seconds: u64,
) -> Result<Value> {
let sql = format!("SELECT count(*) FROM {}", sql_ident(table_name));
let deadline = Instant::now() + Duration::from_secs(visibility_timeout_seconds);
let mut attempts = 0;
loop {
attempts += 1;
let mut result = http_post_sql(client, port, &sql, database).await;
let observed_rows = extract_count_value(&result);
let row_count_ok =
result["ok"].as_bool().unwrap_or(false) && observed_rows == Some(expected_rows);
result
.as_object_mut()
.ok_or("count result must be an object")?
.extend([
("expected_rows".to_string(), json!(expected_rows)),
("observed_rows".to_string(), json!(observed_rows)),
("attempts".to_string(), json!(attempts)),
("row_count_ok".to_string(), json!(row_count_ok)),
]);
if row_count_ok {
return Ok(result);
}
if Instant::now() >= deadline {
return Err(format!(
"expected {expected_rows} rows but observed {observed_rows:?} after {attempts} attempts"
)
.into());
}
tokio::time::sleep(Duration::from_millis(500)).await;
}
}
fn json_number(value: f64) -> String {
serde_json::to_string(&value).unwrap_or_else(|_| value.to_string())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::query_regression_runner::model::{PromStore, RemoteValue};
#[test]
fn schedules_remote_sample_chunks_and_flushes() {
let remote = RemoteWrite {
database: "public".to_string(),
metric: "metric".to_string(),
physical_table: "physical".to_string(),
series_count: 2,
samples_per_series: 5,
start_unix_millis: 100,
step_millis: 10,
chunk_series_count: 1,
timeout_seconds: 60,
sample_chunk_size: Some(2),
flush_every_sample_chunks: 2,
visibility_timeout_seconds: 30,
prom_store: PromStore {
pending_rows_flush_interval: "1s".to_string(),
max_batch_rows: 1,
max_concurrent_flushes: 1,
worker_channel_capacity: 1,
max_inflight_requests: 1,
},
value: RemoteValue {
pattern: "linear".to_string(),
base: 0.0,
step: 1.0,
cardinality: 1,
seed: 0,
run_length: 1,
stall_every: 0,
stall_length: 0,
mixed_every: 0,
},
storage: None,
read_bench: None,
};
let chunks = sample_chunks(&remote).unwrap().unwrap();
assert_eq!(
chunks,
vec![
SampleChunk {
index: 1,
offset: 0,
samples_per_series: 2,
start_unix_millis: 100,
},
SampleChunk {
index: 2,
offset: 2,
samples_per_series: 2,
start_unix_millis: 120,
},
SampleChunk {
index: 3,
offset: 4,
samples_per_series: 1,
start_unix_millis: 140,
},
]
);
assert_eq!(
scheduled_flushes(&chunks, remote.flush_every_sample_chunks),
vec![(2, "periodic"), (3, "final")]
);
}
}
@@ -0,0 +1,183 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::fs;
use std::path::{Path, PathBuf};
use serde_json::{Value, json};
use self::read_bench::run_read_bench;
use self::storage::{enforce_storage_thresholds, run_storage_inspection};
use crate::query_regression_runner::model::DestinationConfig;
use crate::query_regression_runner::plan::normalized_remote_write;
use crate::query_regression_runner::{
FinalizeRemoteArgs, PrepareRemoteArgs, RenderRemoteConfigArgs, Result,
};
mod ingest;
mod read_bench;
mod storage;
pub(super) async fn run_render_remote_config(args: RenderRemoteConfigArgs) -> Result<()> {
ingest::run_render_remote_config(args).await
}
pub(super) async fn run_prepare_remote(args: PrepareRemoteArgs) -> Result<()> {
ingest::run_prepare_remote(args).await
}
pub(super) async fn run_finalize_remote(args: FinalizeRemoteArgs) -> Result<()> {
let (_, remote) = normalized_remote_write(&args.fixture_generator, &args.case)?;
let storage = remote.storage.as_ref().filter(|storage| storage.inspect);
let mut report: Value = serde_json::from_slice(&fs::read(&args.report)?)?;
let bench_root = args.report.parent().unwrap_or_else(|| Path::new("."));
let mut inspections = Vec::new();
let target_failed = {
let targets = report
.get_mut("targets")
.and_then(Value::as_array_mut)
.ok_or("report has no targets array")?;
for (name, data_home, destination) in [
(
"base",
args.base_data_home.as_deref(),
args.base_destination.as_deref(),
),
(
"candidate",
args.candidate_data_home.as_deref(),
args.candidate_destination.as_deref(),
),
] {
let (data_home, destination) = match (data_home, destination) {
(Some(data_home), None) => (data_home.to_path_buf(), None),
(None, Some(path)) => {
let destination: DestinationConfig =
toml::from_str(&fs::read_to_string(path)?)?;
(
PathBuf::from(destination.data_home),
Some(path.to_path_buf()),
)
}
(Some(_), Some(_)) => {
return Err(format!(
"{name}: --{name}-data-home and --{name}-destination are mutually exclusive"
)
.into());
}
(None, None) => {
return Err(format!(
"{name}: one of --{name}-data-home or --{name}-destination is required"
)
.into());
}
};
let target = targets
.iter_mut()
.find(|target| target.get("name").and_then(Value::as_str) == Some(name))
.ok_or_else(|| format!("report has no {name} target"))?;
let Some(storage) = storage else {
target
.as_object_mut()
.ok_or("report target must be an object")?
.insert(
"read_bench".to_string(),
json!({"status": "skipped", "reason": "storage inspection disabled"}),
);
continue;
};
let inspection = run_storage_inspection(
&args.fixture_generator,
&data_home,
destination.as_deref(),
storage,
)?;
let bench_dir = bench_root.join(name).join("read_bench");
let read_bench = run_read_bench(
&args.candidate_bin,
&data_home,
&bench_dir,
remote.read_bench.as_ref(),
&inspection,
)?;
let inspection_failed = inspection["status"] == "failed";
let bench_failed = read_bench["status"] == "failed";
let target = target
.as_object_mut()
.ok_or("report target must be an object")?;
target.insert("storage_inspection".to_string(), inspection.clone());
target.insert("read_bench".to_string(), read_bench);
if inspection_failed || bench_failed {
target.insert("status".to_string(), json!("failed"));
}
inspections.push(inspection);
}
targets.iter().any(|target| {
target["status"] == "failed"
|| target["storage_inspection"]["status"] == "failed"
|| target["read_bench"]["status"] == "failed"
})
};
let storage_thresholds = storage
.map(|storage| {
enforce_storage_thresholds(
storage,
inspections.first().unwrap_or(&Value::Null),
inspections.get(1).unwrap_or(&Value::Null),
)
})
.unwrap_or_default();
let threshold_failed = {
let thresholds = report
.get_mut("thresholds")
.and_then(Value::as_array_mut)
.ok_or("report has no thresholds array")?;
thresholds.retain(|threshold| !is_storage_threshold_entry(threshold));
thresholds.extend(storage_thresholds);
thresholds
.iter()
.any(|threshold| threshold["status"] == "failed")
};
let failed = target_failed || threshold_failed;
report["status"] = json!(if failed { "failed" } else { "ok" });
let text = format!("{}\n", serde_json::to_string_pretty(&report)?);
fs::write(&args.report, &text)?;
print!("{text}");
if failed {
std::process::exit(1);
}
Ok(())
}
fn is_storage_threshold_entry(threshold: &Value) -> bool {
let Some(name) = threshold.get("threshold").and_then(Value::as_str) else {
return false;
};
matches!(
name,
"max_candidate_total_file_size_regression_pct"
| "max_candidate_column_compressed_size_regression_pct"
| "max_candidate_column_uncompressed_size_regression_pct"
) || (threshold.get("target").is_some()
&& matches!(
name,
"min_files"
| "min_files_with_column"
| "max_total_file_size_bytes"
| "max_column_compressed_size_bytes"
| "max_column_uncompressed_size_bytes"
| "require_encodings"
| "forbid_encodings"
))
}
@@ -0,0 +1,363 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::fs;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::Instant;
use regex::Regex;
use serde_json::{Value, json};
use crate::query_regression_runner::Result;
use crate::query_regression_runner::measure::median;
use crate::query_regression_runner::model::ReadBenchConfig;
#[derive(Clone, Debug)]
struct BenchTarget {
relative_path: String,
table_dir: String,
region_id: String,
path_type: String,
file_id: String,
}
pub(super) fn run_read_bench(
candidate_bin: &Path,
data_home: &Path,
bench_dir: &Path,
read_bench: Option<&ReadBenchConfig>,
inspection: &Value,
) -> Result<Value> {
let Some(read_bench) = read_bench.filter(|config| config.enabled) else {
return Ok(json!({ "status": "skipped", "reason": "read_bench disabled" }));
};
if !read_bench.parquetbench && !read_bench.scanbench {
return Ok(json!({ "status": "skipped", "reason": "parquetbench and scanbench disabled" }));
}
let report = inspection
.get("summary")
.filter(|summary| summary.is_object());
let files = report
.and_then(|report| report.get("files"))
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
let root = report
.and_then(|report| report.get("root"))
.and_then(Value::as_str)
.map(PathBuf::from)
.or_else(|| {
inspection
.get("root")
.and_then(Value::as_str)
.map(PathBuf::from)
})
.ok_or("storage inspection has no root")?;
let mut targets = files
.iter()
.filter(|file| {
file.get("relative_path")
.and_then(Value::as_str)
.is_some_and(|path| path.ends_with(".parquet"))
&& file
.get("columns")
.and_then(Value::as_array)
.is_some_and(|columns| !columns.is_empty())
})
.filter_map(|file| inspected_bench_target(data_home, &root, file).transpose())
.collect::<Result<Vec<_>>>()?;
if let Some(max_files) = read_bench.max_files {
targets.truncate(max_files);
}
if targets.is_empty() {
return Ok(
json!({ "status": "failed", "reason": "no inspected data SST files available for read_bench" }),
);
}
fs::create_dir_all(bench_dir)?;
let config_toml = bench_dir.join("bench.toml");
let scan_json = bench_dir.join("scan.json");
fs::write(
&config_toml,
format!(
"[storage]\ndata_home = \"{}\"\ntype = \"File\"\n\n[[region_engine]]\n[region_engine.mito]\n",
data_home.display()
),
)?;
fs::write(
&scan_json,
format!(
"{}\n",
serde_json::to_string_pretty(&json!({ "projection_names": read_bench.projection }))?
),
)?;
let mut parquet_runs = Vec::new();
if read_bench.parquetbench {
for target in &targets {
let command = vec![
candidate_bin.to_string_lossy().to_string(),
"datanode".to_string(),
"parquetbench".to_string(),
"--config".to_string(),
config_toml.to_string_lossy().to_string(),
"--region-id".to_string(),
target.region_id.clone(),
"--table-dir".to_string(),
target.table_dir.clone(),
"--file-id".to_string(),
target.file_id.clone(),
"--scan-config".to_string(),
scan_json.to_string_lossy().to_string(),
"--path-type".to_string(),
target.path_type.clone(),
"--iterations".to_string(),
read_bench.iterations.to_string(),
"--reader".to_string(),
read_bench.parquet_reader.clone(),
];
parquet_runs.push(run_bench_command(command, bench_target_value(target))?);
}
}
let mut scan_runs = Vec::new();
if read_bench.scanbench {
for (table_dir, region_id, path_type, files) in group_scan_paths(&targets) {
let command = vec![
candidate_bin.to_string_lossy().to_string(),
"datanode".to_string(),
"scanbench".to_string(),
"--config".to_string(),
config_toml.to_string_lossy().to_string(),
"--region-id".to_string(),
region_id.clone(),
"--table-dir".to_string(),
table_dir.clone(),
"--scan-config".to_string(),
scan_json.to_string_lossy().to_string(),
"--path-type".to_string(),
path_type.clone(),
"--scanner".to_string(),
read_bench.scan_scanner.clone(),
"--parallelism".to_string(),
read_bench.parallelism.to_string(),
"--iterations".to_string(),
read_bench.iterations.to_string(),
];
scan_runs.push(run_bench_command(command, json!({ "table_dir": table_dir, "region_id": region_id, "path_type": path_type, "files": files }))?);
}
}
let failed = parquet_runs
.iter()
.chain(&scan_runs)
.any(|run| run["status"] == "failed");
Ok(json!({
"status": if failed { "failed" } else { "ok" },
"config_path": config_toml,
"scan_config_path": scan_json,
"parquetbench": parquet_runs,
"scanbench": scan_runs,
"aggregate": {
"parquetbench_median_average_ms": bench_median(&parquet_runs),
"scanbench_median_average_ms": bench_median(&scan_runs),
},
}))
}
fn inspected_bench_target(
data_home: &Path,
root: &Path,
file: &Value,
) -> Result<Option<BenchTarget>> {
let relative_path = file
.get("relative_path")
.and_then(Value::as_str)
.unwrap_or_default();
if relative_path.is_empty() {
return Ok(None);
}
let root_suffix = root.strip_prefix(data_home).map_err(|_| {
format!(
"storage inspection root {} is not under datanode data home {}",
root.display(),
data_home.display()
)
})?;
let full_relative = root_suffix.join(relative_path);
let parts = full_relative
.components()
.map(|component| component.as_os_str().to_string_lossy().to_string())
.collect::<Vec<_>>();
if !parts.last().is_some_and(|part| part.ends_with(".parquet")) {
return Ok(None);
}
let Some((region_index, table_id, region_seq)) =
parts.iter().enumerate().find_map(|(index, part)| {
parse_region_dir(part).map(|(table_id, region_seq)| (index, table_id, region_seq))
})
else {
return Ok(None);
};
if region_index == 0 {
return Ok(None);
}
let file_index = parts.len() - 1;
let path_type = if region_index + 1 < file_index
&& matches!(parts[region_index + 1].as_str(), "data" | "metadata")
{
parts[region_index + 1].clone()
} else {
"bare".to_string()
};
if path_type == "metadata" {
return Ok(None);
}
let file_id = parts[file_index].trim_end_matches(".parquet").to_string();
Ok(Some(BenchTarget {
relative_path: full_relative.to_string_lossy().to_string(),
table_dir: format!("{}/", parts[..region_index].join("/")),
region_id: format!("{table_id}:{region_seq}"),
path_type,
file_id,
}))
}
fn parse_region_dir(part: &str) -> Option<(u64, u64)> {
let (table_id, region_seq) = part.split_once('_')?;
(region_seq.len() == 10
&& table_id.chars().all(|character| character.is_ascii_digit())
&& region_seq
.chars()
.all(|character| character.is_ascii_digit()))
.then(|| Some((table_id.parse().ok()?, region_seq.parse().ok()?)))?
}
fn bench_target_value(target: &BenchTarget) -> Value {
json!({
"relative_path": target.relative_path,
"table_dir": target.table_dir,
"region_id": target.region_id,
"path_type": target.path_type,
"file_id": target.file_id,
})
}
fn group_scan_paths(targets: &[BenchTarget]) -> Vec<(String, String, String, Vec<String>)> {
let mut groups: Vec<(String, String, String, Vec<String>)> = Vec::new();
for target in targets {
if let Some((_, _, _, paths)) =
groups
.iter_mut()
.find(|(table_dir, region_id, path_type, _)| {
table_dir == &target.table_dir
&& region_id == &target.region_id
&& path_type == &target.path_type
})
{
paths.push(target.relative_path.clone());
} else {
groups.push((
target.table_dir.clone(),
target.region_id.clone(),
target.path_type.clone(),
vec![target.relative_path.clone()],
));
}
}
groups
}
fn run_bench_command(command: Vec<String>, mut run: Value) -> Result<Value> {
let started = Instant::now();
let output = Command::new(&command[0]).args(&command[1..]).output()?;
let elapsed_seconds = started.elapsed().as_secs_f64();
let stdout = String::from_utf8_lossy(&output.stdout).to_string();
let stderr = String::from_utf8_lossy(&output.stderr).to_string();
let run = run.as_object_mut().ok_or("bench run must be an object")?;
run.extend([
("command".to_string(), json!(command)),
("returncode".to_string(), json!(output.status.code())),
("elapsed_seconds".to_string(), json!(elapsed_seconds)),
("stdout".to_string(), json!(stdout)),
("stderr".to_string(), json!(stderr)),
(
"status".to_string(),
json!(if output.status.success() {
"ok"
} else {
"failed"
}),
),
]);
if let Some(average_ms) = parse_average_duration(run["stdout"].as_str().unwrap_or_default()) {
run.insert("average_ms".to_string(), json!(average_ms));
}
Ok(Value::Object(run.clone()))
}
fn parse_average_duration(stdout: &str) -> Option<f64> {
Regex::new(r"(?i)Average duration[^0-9]*([0-9.]+)\s*ms")
.ok()?
.captures(stdout)?
.get(1)?
.as_str()
.parse()
.ok()
}
fn bench_median(runs: &[Value]) -> Option<f64> {
let values = runs
.iter()
.filter_map(|run| run.get("average_ms").and_then(Value::as_f64))
.collect::<Vec<_>>();
(!values.is_empty()).then(|| median(&values))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parses_average_duration_and_groups_scan_paths() {
assert_eq!(
parse_average_duration("work\nAverage duration: 12.5 ms\n"),
Some(12.5)
);
assert_eq!(parse_average_duration("completed"), None);
let targets = vec![
BenchTarget {
relative_path: "data/a.parquet".to_string(),
table_dir: "data/".to_string(),
region_id: "1:2".to_string(),
path_type: "data".to_string(),
file_id: "a".to_string(),
},
BenchTarget {
relative_path: "data/b.parquet".to_string(),
table_dir: "data/".to_string(),
region_id: "1:2".to_string(),
path_type: "data".to_string(),
file_id: "b".to_string(),
},
];
assert_eq!(
group_scan_paths(&targets),
vec![(
"data/".to_string(),
"1:2".to_string(),
"data".to_string(),
vec!["data/a.parquet".to_string(), "data/b.parquet".to_string()],
)]
);
}
}
@@ -0,0 +1,241 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::path::Path;
use std::process::Command;
use std::time::Instant;
use serde_json::{Value, json};
use crate::query_regression_runner::Result;
use crate::query_regression_runner::model::StorageConfig;
use crate::query_regression_runner::sql::{value_f64, value_u64};
pub(super) fn run_storage_inspection(
generator: &Path,
data_home: &Path,
destination: Option<&Path>,
storage: &StorageConfig,
) -> Result<Value> {
let mut command = vec![
generator.to_string_lossy().to_string(),
"inspect-footer".to_string(),
"--column".to_string(),
storage.column.clone(),
];
let root = if let Some(destination) = destination {
command.push("--destination".to_string());
command.push(destination.to_string_lossy().to_string());
data_home.to_path_buf()
} else {
let root = storage
.root_suffix
.as_deref()
.map_or_else(|| data_home.to_path_buf(), |suffix| data_home.join(suffix));
command.push("--root".to_string());
command.push(root.to_string_lossy().to_string());
root
};
if storage.include_metadata_files {
command.push("--include-metadata-files".to_string());
}
let started = Instant::now();
let output = Command::new(generator).args(&command[1..]).output()?;
let elapsed_seconds = started.elapsed().as_secs_f64();
let stdout = String::from_utf8_lossy(&output.stdout).to_string();
let stderr = String::from_utf8_lossy(&output.stderr).to_string();
if !output.status.success() {
return Err(format!(
"storage inspector failed: command={command:?}, returncode={:?}, elapsed_seconds={elapsed_seconds:.3}, stderr={:.2000}",
output.status.code(), stderr
)
.into());
}
let mut result = json!({
"status": "ok",
"command": command,
"root": root,
"returncode": output.status.code(),
"elapsed_seconds": elapsed_seconds,
"stdout": stdout,
"stderr": stderr,
});
match serde_json::from_str(result["stdout"].as_str().unwrap_or_default()) {
Ok(summary) => result["summary"] = summary,
Err(_) => {
result["summary_parse_error"] = result["stdout"].clone();
result["status"] = json!("failed");
}
}
Ok(result)
}
fn storage_summary(inspection: &Value) -> Value {
inspection
.get("summary")
.and_then(|summary| summary.get("summary"))
.filter(|summary| summary.is_object())
.cloned()
.unwrap_or_else(|| json!({}))
}
pub(super) fn enforce_storage_thresholds(
storage: &StorageConfig,
base_inspection: &Value,
candidate_inspection: &Value,
) -> Vec<Value> {
let base = storage_summary(base_inspection);
let candidate = storage_summary(candidate_inspection);
let targets = [("base", &base), ("candidate", &candidate)];
let mut results = Vec::new();
for (threshold, field, limit) in [
("min_files", "file_count", storage.min_files),
(
"min_files_with_column",
"files_with_column",
storage.min_files_with_column,
),
] {
for (target, summary) in &targets {
let actual = summary.get(field).cloned().unwrap_or(Value::Null);
let ok = value_u64(Some(&actual)).is_some_and(|actual| actual >= limit);
results.push(json!({ "target": target, "threshold": threshold, "status": if ok { "passed" } else { "failed" }, "actual": actual, "limit": limit }));
}
}
for (threshold, field, limit) in [
(
"max_total_file_size_bytes",
"total_file_size",
storage.max_total_file_size_bytes,
),
(
"max_column_compressed_size_bytes",
"column_compressed_size",
storage.max_column_compressed_size_bytes,
),
(
"max_column_uncompressed_size_bytes",
"column_uncompressed_size",
storage.max_column_uncompressed_size_bytes,
),
] {
let Some(limit) = limit else { continue };
for (target, summary) in &targets {
let actual = summary.get(field).cloned().unwrap_or(Value::Null);
let ok = value_f64(Some(&actual)).is_some_and(|actual| actual <= limit as f64);
results.push(json!({ "target": target, "threshold": threshold, "status": if ok { "passed" } else { "failed" }, "actual": actual, "limit": limit }));
}
}
for (target, summary) in &targets {
let encodings = summary
.get("unique_encodings")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
for encoding in &storage.require_encodings {
results.push(json!({ "target": target, "threshold": "require_encodings", "encoding": encoding, "status": if encodings.iter().any(|value| value.as_str() == Some(encoding)) { "passed" } else { "failed" } }));
}
for encoding in &storage.forbid_encodings {
results.push(json!({ "target": target, "threshold": "forbid_encodings", "encoding": encoding, "status": if encodings.iter().all(|value| value.as_str() != Some(encoding)) { "passed" } else { "failed" } }));
}
}
for (threshold, field, limit) in [
(
"max_candidate_total_file_size_regression_pct",
"total_file_size",
storage.max_candidate_total_file_size_regression_pct,
),
(
"max_candidate_column_compressed_size_regression_pct",
"column_compressed_size",
storage.max_candidate_column_compressed_size_regression_pct,
),
(
"max_candidate_column_uncompressed_size_regression_pct",
"column_uncompressed_size",
storage.max_candidate_column_uncompressed_size_regression_pct,
),
] {
let Some(limit) = limit else { continue };
let base_value = base.get(field).cloned().unwrap_or(Value::Null);
let candidate_value = candidate.get(field).cloned().unwrap_or(Value::Null);
let (Some(base_number), Some(candidate_number)) = (
value_f64(Some(&base_value)).filter(|value| *value != 0.0),
value_f64(Some(&candidate_value)),
) else {
results.push(json!({ "threshold": threshold, "status": "failed", "reason": "missing or zero base/candidate value", "base": base_value, "candidate": candidate_value }));
continue;
};
let actual = (candidate_number - base_number) / base_number * 100.0;
results.push(json!({ "threshold": threshold, "status": if actual <= limit { "passed" } else { "failed" }, "actual_pct": actual, "limit_pct": limit, "base": base_value, "candidate": candidate_value }));
}
results
}
#[cfg(test)]
mod tests {
use super::super::is_storage_threshold_entry;
use super::*;
#[test]
fn enforces_storage_thresholds_for_each_target_and_regression() {
let storage: StorageConfig = serde_json::from_value(json!({
"inspect": true,
"column": "value",
"root_suffix": null,
"include_metadata_files": false,
"min_files": 2,
"min_files_with_column": 1,
"require_encodings": ["PLAIN"],
"forbid_encodings": ["DELTA"],
"max_total_file_size_bytes": 105,
"max_column_compressed_size_bytes": null,
"max_column_uncompressed_size_bytes": null,
"max_candidate_total_file_size_regression_pct": 5.0,
"max_candidate_column_compressed_size_regression_pct": null,
"max_candidate_column_uncompressed_size_regression_pct": null
}))
.unwrap();
let base = json!({"summary": {"summary": {"file_count": 2, "files_with_column": 1, "total_file_size": 100, "unique_encodings": ["PLAIN"]}}});
let candidate = json!({"summary": {"summary": {"file_count": 2, "files_with_column": 1, "total_file_size": 110, "unique_encodings": ["PLAIN"]}}});
let results = enforce_storage_thresholds(&storage, &base, &candidate);
assert!(
results
.iter()
.any(|result| result["threshold"] == "max_total_file_size_bytes"
&& result["target"] == "candidate"
&& result["status"] == "failed")
);
assert!(results.iter().any(|result| result["threshold"]
== "max_candidate_total_file_size_regression_pct"
&& result["actual_pct"] == 10.0
&& result["status"] == "failed"));
assert!(results.iter().any(
|result| result["threshold"] == "require_encodings" && result["status"] == "passed"
));
}
#[test]
fn identifies_prior_storage_threshold_entries() {
assert!(is_storage_threshold_entry(
&json!({"target": "base", "threshold": "min_files"})
));
assert!(is_storage_threshold_entry(&json!({
"threshold": "max_candidate_total_file_size_regression_pct"
})));
assert!(!is_storage_threshold_entry(&json!({
"threshold": "max_candidate_latency_regression_pct"
})));
}
}
@@ -0,0 +1,238 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::time::Instant;
use reqwest::Client;
use serde_json::{Value, json};
use crate::query_regression_runner::Result;
pub(super) fn sql_string(value: &str) -> String {
format!("'{}'", value.replace('\'', "''"))
}
pub(super) fn extract_rows(body: &Value) -> Vec<Value> {
fn visit(body: &Value, rows: &mut Vec<Value>) {
match body {
Value::Object(object) => {
for key in ["data", "rows", "records", "output"] {
let Some(value) = object.get(key) else {
continue;
};
if matches!(key, "data" | "rows") && value.is_array() {
rows.extend(value.as_array().unwrap().iter().cloned());
} else {
visit(value, rows);
}
}
}
Value::Array(values) => {
if !values.is_empty()
&& values
.iter()
.all(|value| !value.is_object() && !value.is_array())
{
rows.push(body.clone());
} else {
for value in values {
visit(value, rows);
}
}
}
_ => {}
}
}
let mut rows = Vec::new();
visit(body, &mut rows);
rows
}
pub(super) fn row_value<'a>(row: &'a Value, index: usize, name: &str) -> Option<&'a Value> {
match row {
Value::Object(values) => [
name.to_string(),
name.to_ascii_uppercase(),
name.to_ascii_lowercase(),
]
.into_iter()
.find_map(|key| values.get(&key)),
Value::Array(values) => values.get(index),
_ => Some(row),
}
}
pub(super) fn row_u64(row: &Value, index: usize, name: &str) -> Result<u64> {
let value =
row_value(row, index, name).ok_or_else(|| format!("missing {name} in row {row}"))?;
value
.as_u64()
.or_else(|| value.as_str().and_then(|value| value.parse().ok()))
.ok_or_else(|| format!("invalid {name} in row {row}").into())
}
pub(super) fn value_text(value: &Value) -> String {
match value {
Value::String(value) => value.clone(),
Value::Bool(value) => value.to_string(),
Value::Number(value) => value.to_string(),
Value::Null => "None".to_string(),
_ => value.to_string(),
}
}
pub(super) fn extract_count_value(result: &Value) -> Option<u64> {
let row = result
.get("response")?
.get("data")?
.as_array()?
.first()?
.as_object()?;
row.iter()
.find(|(key, _)| {
let key = key.to_ascii_lowercase();
key == "count(*)" || key.starts_with("count(")
})
.and_then(|(_, value)| value_u64(Some(value)))
}
pub(super) fn value_u64(value: Option<&Value>) -> Option<u64> {
value?.as_u64().or_else(|| value?.as_str()?.parse().ok())
}
pub(super) fn value_f64(value: Option<&Value>) -> Option<f64> {
value?.as_f64().or_else(|| value?.as_str()?.parse().ok())
}
pub(super) async fn http_post_sql(client: &Client, port: u16, sql: &str, db: &str) -> Value {
let started = Instant::now();
let request = client
.post(format!("http://127.0.0.1:{port}/v1/sql"))
.form(&[("sql", sql), ("db", db), ("format", "json")]);
match request.send().await {
Ok(response) => {
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);
let mut sample = json!({
"ok": ok,
"status": status,
"latency_ms": started.elapsed().as_secs_f64() * 1000.0,
"response": body,
"sql": sql,
});
if status >= 400 {
sample
.as_object_mut()
.expect("HTTP samples are objects")
.insert("error".to_string(), Value::String(format!("HTTP {status}")));
}
sample
}
Err(error) => json!({
"ok": false,
"status": status,
"latency_ms": started.elapsed().as_secs_f64() * 1000.0,
"error": error.to_string(),
"sql": sql,
}),
}
}
Err(error) => json!({
"ok": false,
"status": Value::Null,
"latency_ms": started.elapsed().as_secs_f64() * 1000.0,
"error": error.to_string(),
"sql": sql,
}),
}
}
fn response_has_error(body: &Value) -> bool {
let Some(body) = body.as_object() else {
return false;
};
["error", "err_msg", "error_msg"]
.into_iter()
.any(|key| body.get(key).is_some_and(is_truthy))
|| body
.get("error_code")
.is_some_and(|value| !is_success_code(value))
|| (!body.contains_key("output")
&& body
.get("code")
.is_some_and(|value| !is_success_code(value)))
}
fn is_truthy(value: &Value) -> bool {
match value {
Value::Null => false,
Value::Bool(value) => *value,
Value::Number(value) => value.as_f64().is_none_or(|value| value != 0.0),
Value::String(value) => !value.is_empty(),
Value::Array(value) => !value.is_empty(),
Value::Object(value) => !value.is_empty(),
}
}
fn is_success_code(value: &Value) -> bool {
let code = match value {
Value::String(value) => value.clone(),
Value::Bool(value) => value.to_string(),
Value::Number(value) => value.to_string(),
Value::Null => "None".to_string(),
_ => value.to_string(),
};
matches!(code.to_lowercase().as_str(), "" | "0" | "success")
}
pub(super) fn sql_ident(name: &str) -> String {
format!("\"{}\"", name.replace('"', "\"\""))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn top_level_errors_do_not_inspect_rows() {
assert!(response_has_error(&json!({"error_code": 7})));
assert!(response_has_error(&json!({"code": "bad"})));
assert!(!response_has_error(
&json!({"output": [{"code": 7, "error": "row value"}]})
));
assert!(!response_has_error(
&json!({"error_code": "success", "output": []})
));
}
#[test]
fn extracts_remote_write_count_from_data_map() {
assert_eq!(
extract_count_value(&json!({"response": {"data": [{"COUNT(*)": "12"}]}})),
Some(12)
);
assert_eq!(
extract_count_value(&json!({"response": {"data": [{"count(value)": 7}]}})),
Some(7)
);
assert_eq!(
extract_count_value(&json!({"response": {"data": [[]]}})),
None
);
}
}