test(perf): add default versus BSS storage comparison

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

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<String>,
+    #[serde(default)]
+    pub(super) candidate_setup_sql: Vec<String>,
+    #[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 {

src/cmd/src/bin/query_regression_runner/model.rs

@@ -116,6 +116,10 @@ pub(super) struct RemoteWrite {
     pub(super) sample_chunk_size: Option<u64>,
     pub(super) flush_every_sample_chunks: u64,
     pub(super) visibility_timeout_seconds: u64,
+    #[serde(default)]
+    pub(super) base_setup_sql: Vec<String>,
+    #[serde(default)]
+    pub(super) candidate_setup_sql: Vec<String>,
     pub(super) prom_store: PromStore,
     pub(super) value: RemoteValue,
     pub(super) storage: Option<StorageConfig>,

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<Value> {
     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<Vec<Value>> {
+    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<u8> {
+        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::<usize>()
+                .unwrap();
+            if request.len() >= header_end + 4 + content_length {
+                return request;
+            }
+        }
+    }
+
+    async fn spawn_sql_server(
+        responses: Vec<String>,
+    ) -> (u16, tokio::task::JoinHandle<Vec<String>>) {
+        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,

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

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