diff --git a/src/cmd/src/bin/query_perf_fixture/case.rs b/src/cmd/src/bin/query_perf_fixture/case.rs index a76c49d43d0..693152e6a11 100644 --- a/src/cmd/src/bin/query_perf_fixture/case.rs +++ b/src/cmd/src/bin/query_perf_fixture/case.rs @@ -105,6 +105,10 @@ pub(super) struct PromRemoteWritePlan { #[serde(default = "default_visibility_timeout_seconds")] pub(super) visibility_timeout_seconds: u64, #[serde(default)] + pub(super) base_setup_sql: Vec, + #[serde(default)] + pub(super) candidate_setup_sql: Vec, + #[serde(default)] pub(super) prom_store: PromStoreConfig, #[serde(default)] pub(super) value: ValueConfig, @@ -436,6 +440,63 @@ pub(super) fn default_parallelism() -> u64 { 1 } +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn remote_write_setup_sql_defaults_to_empty() { + let case: CaseFile = toml::from_str( + r#" +[scenario] +kind = "prom_remote_write_then_query" + +[scenario.remote_write] +metric = "metric" +"#, + ) + .unwrap(); + let Scenario::PromRemoteWriteThenQuery(scenario) = case.scenario else { + panic!("expected prom_remote_write_then_query scenario"); + }; + assert!(scenario.remote_write.base_setup_sql.is_empty()); + assert!(scenario.remote_write.candidate_setup_sql.is_empty()); + } + + #[test] + fn remote_write_setup_sql_roundtrips() { + let case: CaseFile = toml::from_str( + r#" +[scenario] +kind = "prom_remote_write_then_query" + +[scenario.remote_write] +metric = "metric" +base_setup_sql = ["CREATE TABLE base_table"] +candidate_setup_sql = ["CREATE TABLE candidate_table", "ALTER TABLE candidate_table SET 'x'='y'"] +"#, + ) + .unwrap(); + + let roundtrip: CaseFile = + serde_json::from_str(&serde_json::to_string(&case).unwrap()).unwrap(); + let Scenario::PromRemoteWriteThenQuery(scenario) = roundtrip.scenario else { + panic!("expected prom_remote_write_then_query scenario"); + }; + assert_eq!( + scenario.remote_write.base_setup_sql, + ["CREATE TABLE base_table"] + ); + assert_eq!( + scenario.remote_write.candidate_setup_sql, + [ + "CREATE TABLE candidate_table", + "ALTER TABLE candidate_table SET 'x'='y'" + ] + ); + } +} + impl Scenario { pub(super) fn kind(&self) -> &'static str { match self { diff --git a/src/cmd/src/bin/query_regression_runner/model.rs b/src/cmd/src/bin/query_regression_runner/model.rs index 6831a5256a4..5fafdb5635b 100644 --- a/src/cmd/src/bin/query_regression_runner/model.rs +++ b/src/cmd/src/bin/query_regression_runner/model.rs @@ -116,6 +116,10 @@ pub(super) struct RemoteWrite { pub(super) sample_chunk_size: Option, pub(super) flush_every_sample_chunks: u64, pub(super) visibility_timeout_seconds: u64, + #[serde(default)] + pub(super) base_setup_sql: Vec, + #[serde(default)] + pub(super) candidate_setup_sql: Vec, pub(super) prom_store: PromStore, pub(super) value: RemoteValue, pub(super) storage: Option, diff --git a/src/cmd/src/bin/query_regression_runner/remote/ingest.rs b/src/cmd/src/bin/query_regression_runner/remote/ingest.rs index f563402afcd..271e27fac42 100644 --- a/src/cmd/src/bin/query_regression_runner/remote/ingest.rs +++ b/src/cmd/src/bin/query_regression_runner/remote/ingest.rs @@ -57,6 +57,7 @@ pub(super) async fn run_prepare_remote(args: PrepareRemoteArgs) -> Result<()> { args.base_http_port, &args.fixture_generator, &remote, + &remote.base_setup_sql, &client, ) .await?; @@ -65,6 +66,7 @@ pub(super) async fn run_prepare_remote(args: PrepareRemoteArgs) -> Result<()> { args.candidate_http_port, &args.fixture_generator, &remote, + &remote.candidate_setup_sql, &client, ) .await?; @@ -88,6 +90,7 @@ async fn prepare_remote_target( port: u16, generator: &Path, remote: &RemoteWrite, + setup_sql: &[String], client: &Client, ) -> Result { let create_database = http_post_sql( @@ -107,6 +110,7 @@ async fn prepare_remote_target( ) .into()); } + let setup = run_setup_sql(client, port, &remote.database, name, setup_sql).await?; let (remote_write, flushes) = ingest_remote_write(generator, port, remote, client).await?; let expected_rows = remote .series_count @@ -124,6 +128,7 @@ async fn prepare_remote_target( Ok(json!({ "name": name, "create_database": create_database, + "setup_sql": setup, "remote_write": remote_write, "flushes": flushes, "visibility": visibility, @@ -131,6 +136,28 @@ async fn prepare_remote_target( })) } +async fn run_setup_sql( + client: &Client, + port: u16, + database: &str, + target: &str, + setup_sql: &[String], +) -> Result> { + let mut results = Vec::with_capacity(setup_sql.len()); + for (index, statement) in setup_sql.iter().enumerate() { + let result = http_post_sql(client, port, statement, database).await; + if !result["ok"].as_bool().unwrap_or(false) { + return Err(format!( + "setup SQL statement {} failed for {target}: {result}", + index + 1 + ) + .into()); + } + results.push(result); + } + Ok(results) +} + #[derive(Clone, Debug, PartialEq, Eq)] struct SampleChunk { index: u64, @@ -434,9 +461,110 @@ fn json_number(value: f64) -> String { #[cfg(test)] mod tests { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::{TcpListener, TcpStream}; + use super::*; use crate::query_regression_runner::model::{PromStore, RemoteValue}; + async fn read_http_request(stream: &mut TcpStream) -> Vec { + let mut request = Vec::new(); + let mut chunk = [0; 1024]; + loop { + let read = stream.read(&mut chunk).await.unwrap(); + assert_ne!( + read, 0, + "HTTP client closed request before sending its body" + ); + request.extend_from_slice(&chunk[..read]); + let Some(header_end) = request.windows(4).position(|window| window == b"\r\n\r\n") + else { + continue; + }; + let headers = std::str::from_utf8(&request[..header_end]).unwrap(); + let content_length = headers + .lines() + .find_map(|line| line.strip_prefix("content-length: ")) + .or_else(|| { + headers + .lines() + .find_map(|line| line.strip_prefix("Content-Length: ")) + }) + .unwrap() + .parse::() + .unwrap(); + if request.len() >= header_end + 4 + content_length { + return request; + } + } + } + + async fn spawn_sql_server( + responses: Vec, + ) -> (u16, tokio::task::JoinHandle>) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let server = tokio::spawn(async move { + let mut requests = Vec::with_capacity(responses.len()); + for body in responses { + let (mut stream, _) = listener.accept().await.unwrap(); + let request = read_http_request(&mut stream).await; + requests.push(String::from_utf8(request).unwrap()); + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\ + Connection: close\r\nContent-Length: {}\r\n\r\n{body}", + body.len() + ); + stream.write_all(response.as_bytes()).await.unwrap(); + } + requests + }); + (port, server) + } + + #[tokio::test] + async fn setup_sql_runs_in_order_with_configured_database() { + let (port, server) = spawn_sql_server(vec!["{\"code\":0}".to_string(); 2]).await; + let client = Client::new(); + let setup = vec![ + "CREATE TABLE physical (ts TIMESTAMP TIME INDEX) ENGINE=metric".to_string(), + "ALTER TABLE physical SET 'compaction.twcs.trigger_file_num'='100'".to_string(), + ]; + + let results = run_setup_sql(&client, port, "perf", "candidate", &setup) + .await + .unwrap(); + + assert_eq!(results.len(), 2); + let requests = server.await.unwrap(); + assert_eq!(requests.len(), 2); + assert!(requests[0].contains("db=perf")); + assert!(requests[0].contains("sql=CREATE")); + assert!(requests[1].contains("db=perf")); + assert!(requests[1].contains("sql=ALTER")); + } + + #[tokio::test] + async fn setup_sql_stops_at_the_failing_statement() { + let (port, server) = + spawn_sql_server(vec!["{\"code\":0}".to_string(), "{\"code\":1}".to_string()]).await; + let client = Client::new(); + let setup = vec![ + "CREATE TABLE physical (ts TIMESTAMP TIME INDEX) ENGINE=metric".to_string(), + "invalid SQL".to_string(), + "ALTER TABLE physical SET 'compaction.twcs.trigger_file_num'='100'".to_string(), + ]; + + let error = run_setup_sql(&client, port, "perf", "base", &setup) + .await + .unwrap_err(); + + assert!(error.to_string().contains("statement 2 failed for base")); + let requests = server.await.unwrap(); + assert_eq!(requests.len(), 2); + assert!(requests[1].contains("sql=invalid+SQL")); + } + #[test] fn schedules_remote_sample_chunks_and_flushes() { let remote = RemoteWrite { @@ -452,6 +580,8 @@ mod tests { sample_chunk_size: Some(2), flush_every_sample_chunks: 2, visibility_timeout_seconds: 30, + base_setup_sql: Vec::new(), + candidate_setup_sql: Vec::new(), prom_store: PromStore { pending_rows_flush_interval: "1s".to_string(), max_batch_rows: 1, diff --git a/tests/perf/README.md b/tests/perf/README.md index 47495ca531f..2258d484a59 100644 --- a/tests/perf/README.md +++ b/tests/perf/README.md @@ -162,12 +162,15 @@ measures region scan cost, and query measurements still exercise the SQL/TQL frontend path. Treat all performance conclusions as release-only; debug builds are suitable only for command wiring and correctness checks. -`prepare-remote` creates the configured database if needed. The outer driver writes a per-target -frontend config enabling `[prom_store]` with metric engine storage and a non-zero -`pending_rows_flush_interval`, and validates that the logical metric table reaches -`series_count * samples_per_series` rows before trusting the query measurements. -Use `--fixture-generator /path/to/query_perf_fixture` to provide the Rust helper -to the outer driver. +`prepare-remote` creates the configured database if needed. A remote-write case +can provide `base_setup_sql` and `candidate_setup_sql` lists; each target runs its +own complete statements in order after database creation and before ingestion. +The outer driver writes a per-target frontend config enabling `[prom_store]` with +metric engine storage and a non-zero `pending_rows_flush_interval`, and validates +that the logical metric table reaches `series_count * samples_per_series` rows +before trusting the query measurements. Use +`--fixture-generator /path/to/query_perf_fixture` to provide the Rust helper to +the outer driver. Large manual remote-write cases can set `sample_chunk_size` to split ingestion by time. For each chunk, `prepare-remote` invokes `query_perf_fixture prom-remote-write` with the @@ -366,6 +369,57 @@ uv run --no-project python .github/scripts/query-regression-run.py \ --work-dir /tmp/query-regression-work ``` +### Explicit SST float BSS comparison + +`tests/perf/query_cases/sst_float_bss/case.toml` is an explicit-only case for +comparing a default empty physical metric table with a candidate +byte-stream-split (BSS) physical table. Both targets must use the exact same +release `greptime` binary; only the per-target table setup SQL differs. Run the +existing driver from a checkout containing that binary, with absolute paths and +fresh data directories for every run: + +```bash +REPO="$(pwd -P)" +RELEASE_GREPTIME="/absolute/path/to/release/greptime" +FIXTURE_GENERATOR="/absolute/path/to/release/query_perf_fixture" +RUNNER="/absolute/path/to/release/query_regression_runner" +WORK_DIR="/absolute/path/to/fresh/sst-float-bss-run-1" +cd "$REPO" +uv run --no-project python "$REPO/.github/scripts/query-regression-run.py" \ + --cases "$REPO/tests/perf/query_cases/sst_float_bss/case.toml" \ + --base-src "$REPO" \ + --candidate-src . \ + --base-bin "$RELEASE_GREPTIME" \ + --candidate-bin "$RELEASE_GREPTIME" \ + --fixture-generator "$FIXTURE_GENERATOR" \ + --runner "$RUNNER" \ + --work-dir "$WORK_DIR" \ + --summary-script "$REPO/.github/scripts/query-regression-summary.py" +``` + +Repeat the command three times with a different fresh absolute `WORK_DIR` each +run. In each `query-regression-report.json`, compare base and candidate +`targets[].storage_inspection.summary.summary.total_file_size`, and each query's +`targets[].measurements[].latency_ms_median`. Storage percentage is +`(candidate_total_file_size - base_total_file_size) / base_total_file_size * 100`; +query latency percentage is +`(candidate_latency_ms_median - base_latency_ms_median) / base_latency_ms_median * 100`. +The configured three warmups occur after the initial query validation and before +that query's 15 measured endpoint requests, so query latency is a warmed +frontend/cache measurement. After the datanodes stop, the case also runs seven +iterations of `parquetbench` for up to four inspected SST files and `scanbench` +for their scan groups. Their per-run output and aggregate +`parquetbench_median_average_ms` and `scanbench_median_average_ms` are under +`targets[].read_bench`; they are quiescent local-file read/scan diagnostics, not +warmed frontend-query latency measurements. + +The case's `-5.0` storage target and `25` query-latency guardrail are experimental +acceptance targets, not observed-benefit claims; do not relax them if a run +fails. Its three periodic flushes and shared high TWCS trigger avoid the normal +four-file compaction trigger from confounding the layout. Check footer encodings +(BSS on candidate and no BSS on base) and data equality from the artifacts rather +than through a new harness framework. + The Rust runner subcommands are also useful for focused diagnostics: ```bash diff --git a/tests/perf/query_cases/sst_float_bss/case.toml b/tests/perf/query_cases/sst_float_bss/case.toml new file mode 100644 index 00000000000..1e3f733462f --- /dev/null +++ b/tests/perf/query_cases/sst_float_bss/case.toml @@ -0,0 +1,74 @@ +# Explicit-only default-versus-BSS storage and warmed-query comparison. This is +# intentionally outside the routine default case set. + +[case] +name = "sst_float_bss" +description = "Compare default and byte-stream-split floating-point SST bytes and warmed value-scanning query latency" + +[scenario] +kind = "prom_remote_write_then_query" + +[scenario.remote_write] +database = "public" +metric = "sst_float_bss" +physical_table = "sst_float_bss_physical" +series_count = 1024 +samples_per_series = 4320 +sample_chunk_size = 1440 +flush_every_sample_chunks = 1 +start_unix_millis = 1_704_067_200_000 # 2024-01-01T00:00:00Z +step_millis = 60000 +chunk_series_count = 128 +timeout_seconds = 600 +visibility_timeout_seconds = 300 + +# Keep the physical-table definition identical except for float encoding. The +# high shared TWCS trigger avoids compaction changing the three-flush layout. +base_setup_sql = [ + "CREATE TABLE sst_float_bss_physical (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ('physical_metric_table'='', 'experimental_sst_float_field_encoding'='default', 'compaction.twcs.trigger_file_num'='100')", +] +candidate_setup_sql = [ + "CREATE TABLE sst_float_bss_physical (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ('physical_metric_table'='', 'experimental_sst_float_field_encoding'='byte_stream_split', 'compaction.twcs.trigger_file_num'='100')", +] + +[scenario.remote_write.value] +pattern = "unique" +base = 0 +step = 1 + +[scenario.remote_write.prom_store] +pending_rows_flush_interval = "1s" +max_batch_rows = 100000 + +[scenario.remote_write.storage] +column = "greptime_value" +min_files = 1 +min_files_with_column = 1 +max_candidate_total_file_size_regression_pct = -5.0 + +[scenario.remote_write.read_bench] +enabled = true +parquetbench = true +scanbench = true +iterations = 7 +max_files = 4 + +[[scenario.queries]] +name = "sum_all_values" +kind = "sql" +query = "SELECT sum(greptime_value) FROM sst_float_bss" +warmup = 3 +iterations = 15 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 25 + +[[scenario.queries]] +name = "hourly_sum_values" +kind = "sql" +query = "SELECT date_bin(INTERVAL '1 hour', greptime_timestamp) AS time_window, sum(greptime_value) FROM sst_float_bss WHERE greptime_timestamp >= TIMESTAMP '2024-01-01 00:00:00' AND greptime_timestamp < TIMESTAMP '2024-01-04 00:00:00' GROUP BY time_window ORDER BY time_window" +warmup = 3 +iterations = 15 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 25