feat: add Prom remote-write query regression scenario (#8413)

* feat: add Prom remote-write query regression scenario Signed-off-by: discord9 <discord9@163.com> * test: add high-cardinality remote-write query case Signed-off-by: discord9 <discord9@163.com> * feat: chunk remote-write query regression loads Signed-off-by: discord9 <discord9@163.com> * test: use multi-day remote-write regression case Signed-off-by: discord9 <discord9@163.com> * fix: address query regression review comments Signed-off-by: discord9 <discord9@163.com> * ci: allow large query regression comments Signed-off-by: discord9 <discord9@163.com> --------- Signed-off-by: discord9 <discord9@163.com>
2026-07-08 23:10:39 +00:00 · 2026-07-07 15:58:56 +08:00
parent 8f5d681e4b
commit 12f83828b1
13 changed files with 791 additions and 75 deletions
--- a/.github/scripts/query-regression-comment.cjs
+++ b/.github/scripts/query-regression-comment.cjs
@@ -49,7 +49,7 @@ function text(value) {
    .replace(/@/g, '@\u200b')
    .replace(/\|/g, '\\|')
    .replace(/\r?\n/g, ' ');
-  return result.length > 512 ? `${result.slice(0, 509)}...` : result;
+  return result;
 }

 function statusEmoji(status) {
@@ -251,10 +251,6 @@ module.exports = async function validateQueryRegressionComment({ github, context
  } else {
    const rendered = [];
    for (const reportPath of reportPaths) {
-      const size = fs.statSync(reportPath).size;
-      if (size > 1024 * 1024) {
-        return skip(core, `Report file too large: ${reportPath} (${size} bytes).`);
-      }
      let report;
      try {
        report = JSON.parse(fs.readFileSync(reportPath, 'utf8'));
@@ -266,9 +262,6 @@ module.exports = async function validateQueryRegressionComment({ github, context
    body += rendered.join('\n\n---\n\n') + '\n';
  }

-  if (body.length > 59999) {
-    body = `${body.slice(0, 59500)}\n\n…\n\n_Comment truncated because it exceeded 60000 characters._\n`;
-  }
  fs.writeFileSync(summaryPath, body);

  core.setOutput('should_post', 'true');
--- a/.github/scripts/query-regression-run.py
+++ b/.github/scripts/query-regression-run.py
@@ -106,6 +106,8 @@ def run_case(args: argparse.Namespace, case_path: Path, work_dir: Path) -> int:
        str(candidate_bin),
        "--fixture-generator",
        str(fixture_generator),
+        "--remote-write-generator",
+        str(args.remote_write_generator),
        "--work-dir",
        str(work_dir),
        "--http-timeout",
@@ -158,6 +160,11 @@ def main() -> int:
    parser.add_argument("--work-dir", default=Path("query-regression-work"), type=Path)
    parser.add_argument("--http-timeout", default=os.environ.get("HTTP_TIMEOUT", "300"))
    parser.add_argument("--allow-large-fixture", default=os.environ.get("ALLOW_LARGE_FIXTURE", "false"))
+    parser.add_argument(
+        "--remote-write-generator",
+        type=Path,
+        default=configured_path(os.environ.get("REMOTE_WRITE_GENERATOR")),
+    )
    parser.add_argument(
        "--summary-script",
        type=Path,
@@ -170,6 +177,8 @@ def main() -> int:
    parser.add_argument("--candidate-ref", default=os.environ.get("CANDIDATE_REF", ""))
    parser.add_argument("--github-output", default=os.environ.get("GITHUB_OUTPUT"))
    args = parser.parse_args()
+    if args.remote_write_generator is None:
+        args.remote_write_generator = args.candidate_src / "target" / profile_dir(args.cargo_profile) / "prom_remote_write_fixture"

    try:
        cases = split_cases(args.cases or [os.environ.get("CASE_PATHS", "all")])
--- a/.github/workflows/query-regression-comment.yml
+++ b/.github/workflows/query-regression-comment.yml
@@ -46,12 +46,6 @@ jobs:
              core.setOutput('found', 'false');
              return;
            }
-            const maxBytes = 1024 * 1024;
-            if (artifact.size_in_bytes > maxBytes) {
-              core.warning(`Comment artifact too large: ${artifact.size_in_bytes} bytes`);
-              core.setOutput('found', 'false');
-              return;
-            }
            core.setOutput('found', 'true');
            core.setOutput('id', String(artifact.id));

--- a/.github/workflows/query-regression.yml
+++ b/.github/workflows/query-regression.yml
@@ -119,10 +119,10 @@ jobs:
          git reset --hard FETCH_HEAD
          git clean -ffdx

-      - name: Build candidate greptime and fixture generator
+      - name: Build candidate greptime and fixture generators
        working-directory: src
        run: |
-          cargo build --profile "${CARGO_PROFILE}" -p cmd --bin greptime --bin query_perf_fixture
+          cargo build --profile "${CARGO_PROFILE}" -p cmd --bin greptime --bin query_perf_fixture --bin prom_remote_write_fixture
          target_dir="${CARGO_PROFILE}"
          if [[ "${CARGO_PROFILE}" == "dev" ]]; then
            target_dir="debug"
@@ -132,6 +132,8 @@ jobs:
            "${GITHUB_WORKSPACE}/query-regression-bins/candidate/greptime"
          cp "${CARGO_TARGET_DIR}/${target_dir}/query_perf_fixture" \
            "${GITHUB_WORKSPACE}/query-regression-bins/candidate/query_perf_fixture"
+          cp "${CARGO_TARGET_DIR}/${target_dir}/prom_remote_write_fixture" \
+            "${GITHUB_WORKSPACE}/query-regression-bins/candidate/prom_remote_write_fixture"

      - name: Run query regression
        id: run
@@ -144,6 +146,7 @@ jobs:
          BASE_BIN: ${{ github.workspace }}/query-regression-bins/base/greptime
          CANDIDATE_BIN: ${{ github.workspace }}/query-regression-bins/candidate/greptime
          FIXTURE_GENERATOR: ${{ github.workspace }}/query-regression-bins/candidate/query_perf_fixture
+          REMOTE_WRITE_GENERATOR: ${{ github.workspace }}/query-regression-bins/candidate/prom_remote_write_fixture
        run: >-
          uv run --no-project python src/.github/scripts/query-regression-run.py
          --base-src src
--- a/src/cmd/Cargo.toml
+++ b/src/cmd/Cargo.toml
@@ -13,6 +13,10 @@ path = "src/bin/greptime.rs"
 name = "query_perf_fixture"
 path = "src/bin/query_perf_fixture.rs"

+[[bin]]
+name = "prom_remote_write_fixture"
+path = "src/bin/prom_remote_write_fixture.rs"
+
 [features]
 default = [
    "servers/pprof",
--- a/src/cmd/src/bin/prom_remote_write_fixture.rs
+++ b/src/cmd/src/bin/prom_remote_write_fixture.rs
@@ -0,0 +1,144 @@
+// 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_stdout)]
+
+use std::time::{Duration, Instant};
+
+use api::prom_store::remote::{Label, Sample, TimeSeries, WriteRequest};
+use clap::Parser;
+use prost::Message;
+use serde_json::json;
+use servers::prom_store::snappy_compress;
+
+#[derive(Debug, Parser)]
+#[command(about = "Generate a deterministic Prometheus remote-write fixture")]
+struct Args {
+    #[arg(long, default_value = "http://127.0.0.1:4000/v1/prometheus/write")]
+    endpoint: String,
+    #[arg(long, default_value = "public")]
+    database: String,
+    #[arg(long)]
+    metric: String,
+    #[arg(long, default_value = "greptime_physical_table")]
+    physical_table: String,
+    #[arg(long, default_value_t = 8)]
+    series_count: u64,
+    #[arg(long, default_value_t = 30)]
+    samples_per_series: u64,
+    #[arg(long, default_value_t = 1_704_067_200_000)]
+    start_unix_millis: i64,
+    #[arg(long, default_value_t = 15_000)]
+    step_millis: i64,
+    #[arg(long, alias = "batch-size", default_value_t = 8)]
+    chunk_series_count: u64,
+    #[arg(long, default_value_t = 60)]
+    timeout_seconds: u64,
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let args = Args::parse();
+    let started = Instant::now();
+    let client = reqwest::Client::builder()
+        .timeout(Duration::from_secs(args.timeout_seconds))
+        .build()?;
+    let mut batches = 0_u64;
+    let mut rows = 0_u64;
+    let mut http_statuses = Vec::new();
+
+    let chunk = args.chunk_series_count.max(1);
+    for first in (0..args.series_count).step_by(chunk as usize) {
+        let last = (first + chunk).min(args.series_count);
+        let timeseries = (first..last)
+            .map(|series_idx| make_series(&args, series_idx))
+            .collect::<Vec<_>>();
+        rows += timeseries
+            .iter()
+            .map(|ts| ts.samples.len() as u64)
+            .sum::<u64>();
+        let body = snappy_compress(
+            &WriteRequest {
+                timeseries,
+                ..Default::default()
+            }
+            .encode_to_vec(),
+        )?;
+        let response = client
+            .post(&args.endpoint)
+            .header("Content-Encoding", "snappy")
+            .body(body)
+            .send()
+            .await?;
+        let status = response.status();
+        http_statuses.push(status.as_u16());
+        if !status.is_success() {
+            let body_text = response.text().await.unwrap_or_default();
+            return Err(format!("remote-write failed with status {status}: {body_text}").into());
+        }
+        batches += 1;
+    }
+
+    println!(
+        "{}",
+        json!({
+            "status": "ok",
+            "endpoint": args.endpoint,
+            "database": args.database,
+            "metric": args.metric,
+            "physical_table": args.physical_table,
+            "series_count": args.series_count,
+            "samples_per_series": args.samples_per_series,
+            "rows": rows,
+            "samples_written": rows,
+            "batches": batches,
+            "elapsed_seconds": started.elapsed().as_secs_f64(),
+            "http_statuses": http_statuses,
+        })
+    );
+    Ok(())
+}
+
+fn make_series(args: &Args, series_idx: u64) -> TimeSeries {
+    let mut labels = vec![
+        label("__name__", &args.metric),
+        label("x_greptime_database", &args.database),
+        label("x_greptime_physical_table", &args.physical_table),
+        label("host", &format!("host{:04}", series_idx % 1024)),
+        label("instance", &format!("instance{:06}", series_idx)),
+    ];
+    labels.sort_unstable_by(|left, right| left.name.cmp(&right.name));
+    let samples = (0..args.samples_per_series)
+        .map(|sample_idx| Sample {
+            value: deterministic_value(series_idx, sample_idx),
+            timestamp: args.start_unix_millis + (sample_idx as i64 * args.step_millis),
+        })
+        .collect();
+    TimeSeries {
+        labels,
+        samples,
+        ..Default::default()
+    }
+}
+
+fn label(name: &str, value: &str) -> Label {
+    Label {
+        name: name.to_string(),
+        value: value.to_string(),
+    }
+}
+
+fn deterministic_value(series_idx: u64, sample_idx: u64) -> f64 {
+    ((series_idx % 97) as f64) + (sample_idx as f64 * 0.125)
+}
--- a/src/cmd/src/bin/query_perf_fixture.rs
+++ b/src/cmd/src/bin/query_perf_fixture.rs
@@ -314,6 +314,25 @@ fn tag_value(col: &ColumnConfig, series: usize) -> String {
    }
 }

+fn series_for_row(layout: &LayoutConfig, sst_idx: usize, base_row: usize, row: usize) -> usize {
+    match layout.series_layout.as_str() {
+        "round_robin" | "timestamp_major" => (base_row + row) % layout.series_count.get(),
+        "per_sst" => sst_idx % layout.series_count.get(),
+        other => panic!("unsupported series_layout {other}"),
+    }
+}
+
+fn timestamp_for_row(layout: &LayoutConfig, base_row: usize, row: usize) -> i64 {
+    let logical_row = base_row + row;
+    let timestamp_step = if layout.series_layout == "timestamp_major" {
+        logical_row / layout.series_count.get()
+    } else {
+        logical_row
+    };
+
+    layout.start_unix_nanos + (timestamp_step as i64 * layout.step_nanos)
+}
+
 fn wave_value(min: f64, max: f64, row: usize) -> f64 {
    let span = max - min;
    let phase = (row % 1024) as f64 / 1023.0;
@@ -337,11 +356,7 @@ fn generate_record_batch(
            ("tag", "STRING") => {
                let mut b = StringDictionaryBuilder::<UInt32Type>::new();
                for row in 0..rows {
-                    let series = if layout.series_layout == "round_robin" {
-                        (base_row + row) % layout.series_count.get()
-                    } else {
-                        sst_idx % layout.series_count.get()
-                    };
+                    let series = series_for_row(layout, sst_idx, base_row, row);
                    let value = tag_value(col, series);
                    b.append_value(value);
                }
@@ -363,16 +378,13 @@ fn generate_record_batch(
            )) as ArrayRef),
            ("timestamp", "TIMESTAMP(9)") => columns.push(Arc::new(TimestampNanosecondArray::from(
                (0..rows)
-                    .map(|r| layout.start_unix_nanos + ((base_row + r) as i64 * layout.step_nanos))
+                    .map(|r| timestamp_for_row(layout, base_row, r))
                    .collect::<Vec<_>>(),
            )) as ArrayRef),
            ("timestamp", "TIMESTAMP(3)") => {
                columns.push(Arc::new(TimestampMillisecondArray::from(
                    (0..rows)
-                        .map(|r| {
-                            (layout.start_unix_nanos + ((base_row + r) as i64 * layout.step_nanos))
-                                / 1_000_000
-                        })
+                        .map(|r| timestamp_for_row(layout, base_row, r) / 1_000_000)
                        .collect::<Vec<_>>(),
                )) as ArrayRef)
            }
@@ -382,11 +394,7 @@ fn generate_record_batch(

    let mut pk_builder = BinaryDictionaryBuilder::<UInt32Type>::new();
    for row in 0..rows {
-        let series = if layout.series_layout == "round_robin" {
-            (base_row + row) % layout.series_count.get()
-        } else {
-            sst_idx % layout.series_count.get()
-        };
+        let series = series_for_row(layout, sst_idx, base_row, row);
        let values = table
            .columns
            .iter()
@@ -493,6 +501,16 @@ async fn main() {
    if scenario.layout.time_range_layout != "non_overlapping_per_sst" {
        panic!("MVP supports time_range_layout=non_overlapping_per_sst");
    }
+    if scenario.layout.series_layout == "timestamp_major"
+        && !scenario
+            .layout
+            .rows_per_sst
+            .is_multiple_of(scenario.layout.series_count.get())
+    {
+        panic!(
+            "series_layout=timestamp_major requires rows_per_sst to be divisible by series_count"
+        );
+    }
    if scenario.layout.sst_count > 1000 && !args.allow_large {
        panic!("sst_count exceeds 1000; pass --allow-large");
    }
--- a/tests/perf/README.md
+++ b/tests/perf/README.md
@@ -34,6 +34,67 @@ counts, time ranges, row groups, and label distributions without spending CI tim
 on ingestion and flush. Ingestion-path cases can be added later for nightly or
 release-level realism.

+## Prometheus remote-write scenario
+
+The runner also supports `scenario.kind = "prom_remote_write_then_query"` for a
+bounded write-path smoke/regression flow. This path is explicit and separate from
+the direct-SST fixture path: it starts the base and candidate distributed clusters,
+writes deterministic Prometheus remote-write v1 samples through
+`/v1/prometheus/write`, flushes the configured physical metric table, then runs
+the configured SQL/TQL queries against the logical metric table.
+
+Remote-write cases configure one database, one logical metric, and one physical
+table under `[scenario.remote_write]`:
+
+```toml
+[scenario]
+kind = "prom_remote_write_then_query"
+
+[scenario.remote_write]
+database = "public"
+metric = "prom_remote_write_smoke"
+physical_table = "greptime_physical_table"
+series_count = 8
+samples_per_series = 30
+start_unix_millis = 1_704_067_200_000
+step_millis = 15_000
+chunk_series_count = 4
+visibility_timeout_seconds = 30
+# Optional: split by time so each helper invocation writes only this many samples
+# per series, then periodically flush the physical metric table to produce
+# multiple time-interval SSTs.
+# sample_chunk_size = 300
+# flush_every_sample_chunks = 1
+
+[scenario.remote_write.prom_store]
+pending_rows_flush_interval = "1s"
+```
+
+The runner creates the configured database if needed, 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 `--remote-write-generator /path/to/prom_remote_write_fixture` to provide the
+Rust payload helper. `--fixture-only` is rejected for remote-write cases; use
+`--dry-run` for planning.
+
+Large manual remote-write cases can set `sample_chunk_size` to split ingestion by
+time. For each chunk, the runner invokes `prom_remote_write_fixture` with the
+same series cardinality but a shorter `--samples-per-series` and an advanced
+`--start-unix-millis`. `flush_every_sample_chunks` controls periodic
+`ADMIN FLUSH_TABLE('<physical_table>')` calls; with `flush_every_sample_chunks = 1`,
+each time chunk is flushed separately. The final visible SST/file-range layout is
+still determined by the storage engine's normal compaction policy, so cases that
+need multi-window file distribution should span multiple compaction windows. If
+`sample_chunk_size` is omitted, the runner keeps the older single-helper-invocation
+behavior and flushes once at the end.
+
+`tests/perf/query_cases/prom_remote_write_7913/case.toml` is a larger manual
+case for issue #7913. It writes 8192 series × 20160 samples through remote-write
+in 1440-sample daily time chunks, flushing after each chunk before running 1d/7d/14d
+TQL selectors. It is not included in the default case set because ingestion cost
+dominates routine CI validation.
+
 ## Generator contract

 The direct-SST generator should accept a case definition with:
@@ -86,9 +147,8 @@ seed = 12345
 # query, warmups, iterations, thresholds
 ```

-The runner currently supports only `direct_readable_sst`. The enum-style
-`scenario.kind` is reserved for future scenarios such as `write_then_query` and
-`cache_warm_query`.
+The runner currently supports `direct_readable_sst` and
+`prom_remote_write_then_query`.

 ## Metrics

@@ -157,6 +217,7 @@ uv run --no-project python tests/perf/query_regression_runner.py \
  --base-bin /path/to/base/greptime \
  --candidate-bin /path/to/candidate/greptime \
  --fixture-generator /path/to/query_perf_fixture \
+  --fixture-cache-dir /mnt/query-regression-fixtures \
  --allow-large-fixture \
  --work-dir /tmp/query-perf-work
 ```
@@ -171,6 +232,17 @@ when intentionally debugging an existing run directory. SQL HTTP requests defaul
 to a 120 second timeout; override with `--http-timeout <seconds>` for slow lab
 runs.

+For large direct-SST fixtures, pass `--fixture-cache-dir <dir>` to store generated
+fixtures in a persistent content-addressed cache keyed by case name and fixture
+data configuration. Query and threshold edits reuse the same cached data as long
+as the scenario layout and table definitions do not change. Cached fixtures are
+reused automatically when their `summary.json`
+matches the discovered table/region metadata; incompatible entries are
+regenerated instead of reused. Fixture materialization keeps base and candidate
+data directories isolated, but copies files efficiently by trying filesystem
+reflinks first, hardlinks for immutable SST/object files next, and normal copies
+as a fallback. Manifest files are reflinked or copied, not hardlinked.
+
 Fixture generator smoke test:

 ```bash
@@ -195,6 +267,18 @@ uv run --no-project python tests/perf/query_regression_runner.py \
 standalone servers, and it materializes the generated fixture into base and
 candidate data directories for plumbing validation.

+Remote-write runner dry-run:
+
+```bash
+uv run --no-project python tests/perf/query_regression_runner.py \
+  --case tests/perf/query_cases/prom_remote_write_smoke/case.toml \
+  --base-bin /path/to/base/greptime \
+  --candidate-bin /path/to/candidate/greptime \
+  --remote-write-generator /path/to/prom_remote_write_fixture \
+  --work-dir /tmp/query-perf-remote-write \
+  --dry-run
+```
+
 ## GitHub Actions

 `.github/workflows/query-regression.yml` provides an opt-in CI entrypoint for
@@ -217,11 +301,13 @@ comment through the trusted follow-up workflow.
 ## Built-in cases

 The `promql_pushdown_7913` case is only one case using the generic fixture
-format. It should generate a metric-like table with a nanosecond time index and
-many SSTs with non-overlapping time ranges. A narrow PromQL/TQL query should
-touch only a small time window. The candidate build is expected to scan
-materially fewer files/ranges/rows than the base build when an optimizer PR
-claims to improve this path.
+format. It generates a high-cardinality metric-like table with a nanosecond time
+index and many SSTs with non-overlapping time ranges. Its `timestamp_major`
+series layout writes one sample for every series at each scrape timestamp, so
+short PromQL/TQL selector windows still scan realistic raw sample volumes. The
+queries should show scan-level time filters, tight SST pruning, and enough raw
+rows to make distributed PromQL pipeline placement meaningful instead of a
+millisecond-scale canary.

 Additional SQL optimizer cases:

--- a/tests/perf/fixture-format.md
+++ b/tests/perf/fixture-format.md
@@ -58,6 +58,13 @@ warmup = 0
 iterations = 1
 ```

+`series_layout = "round_robin"` advances the timestamp once per generated row and
+cycles series labels across rows. `series_layout = "timestamp_major"` writes all
+series for one timestamp before advancing to the next timestamp; use it for
+Prometheus-like high-cardinality scrape fixtures where short query windows should
+still contain many raw samples. `timestamp_major` requires `rows_per_sst` to be
+divisible by `series_count`.
+
 `[scenario]` is required. Other scenario variants are intentionally unsupported
 for now, but `scenario.kind` leaves room for future `write_then_query` and
 `cache_warm_query` configuration.
--- a/tests/perf/query_cases/prom_remote_write_7913/case.toml
+++ b/tests/perf/query_cases/prom_remote_write_7913/case.toml
@@ -0,0 +1,61 @@
+# Manual high-cardinality Prometheus remote-write scenario for validating issue #7913
+# through the normal metric write path. This case is intentionally not part of the
+# default query-regression set; run it explicitly when validating PromQL pipeline
+# placement on realistic metric-engine data.
+
+[case]
+name = "prom_remote_write_7913"
+description = "Multi-day high-cardinality Prometheus remote-write ingestion followed by PromQL/TQL selectors"
+issue = "https://github.com/GreptimeTeam/greptimedb/issues/7913"
+
+[scenario]
+kind = "prom_remote_write_then_query"
+
+[scenario.remote_write]
+database = "public"
+metric = "prom_remote_write_7913"
+physical_table = "greptime_physical_table"
+series_count = 8192
+samples_per_series = 20160
+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 = 512
+timeout_seconds = 600
+visibility_timeout_seconds = 300
+
+[scenario.remote_write.prom_store]
+pending_rows_flush_interval = "1s"
+max_batch_rows = 1_000_000
+max_concurrent_flushes = 256
+
+[[scenario.queries]]
+name = "selector_1d_remote_write_high_cardinality"
+kind = "tql"
+query = "TQL ANALYZE VERBOSE (1704067200, 1704153600, '5m') prom_remote_write_7913{host=~'host.*'}"
+warmup = 1
+iterations = 2
+
+[scenario.queries.thresholds]
+max_candidate_latency_regression_pct = 100
+
+[[scenario.queries]]
+name = "selector_7d_remote_write_high_cardinality"
+kind = "tql"
+query = "TQL ANALYZE VERBOSE (1704067200, 1704672000, '1h') prom_remote_write_7913{host=~'host.*'}"
+warmup = 1
+iterations = 2
+
+[scenario.queries.thresholds]
+max_candidate_latency_regression_pct = 100
+
+[[scenario.queries]]
+name = "selector_14d_remote_write_high_cardinality"
+kind = "tql"
+query = "TQL ANALYZE VERBOSE (1704067200, 1705276800, '1h') prom_remote_write_7913{host=~'host.*'}"
+warmup = 1
+iterations = 2
+
+[scenario.queries.thresholds]
+max_candidate_latency_regression_pct = 100
--- a/tests/perf/query_cases/prom_remote_write_smoke/case.toml
+++ b/tests/perf/query_cases/prom_remote_write_smoke/case.toml
@@ -0,0 +1,31 @@
+[case]
+name = "prom_remote_write_smoke"
+description = "Tiny Prometheus remote-write ingestion then TQL query smoke case"
+
+[scenario]
+kind = "prom_remote_write_then_query"
+
+[scenario.remote_write]
+database = "public"
+metric = "prom_remote_write_smoke"
+physical_table = "greptime_physical_table"
+series_count = 8
+samples_per_series = 30
+start_unix_millis = 1_704_067_200_000
+step_millis = 15_000
+chunk_series_count = 4
+timeout_seconds = 30
+
+[scenario.remote_write.prom_store]
+pending_rows_flush_interval = "1s"
+max_batch_rows = 10000
+
+[[scenario.queries]]
+name = "selector_5m_smoke"
+kind = "tql"
+query = "TQL ANALYZE VERBOSE (1704067200, 1704067650, '15s') prom_remote_write_smoke{host=~'host.*'}"
+warmup = 0
+iterations = 2
+
+[scenario.queries.thresholds]
+max_candidate_latency_regression_pct = 100
--- a/tests/perf/query_cases/promql_pushdown_7913/case.toml
+++ b/tests/perf/query_cases/promql_pushdown_7913/case.toml
@@ -5,7 +5,7 @@

 [case]
 name = "promql_pushdown_7913"
-description = "PromQL/TQL non-ms time-index query should prune SSTs for narrow time ranges"
+description = "PromQL/TQL small-window selectors should stay fast on high-cardinality metric-like fixtures"
 issue = "https://github.com/GreptimeTeam/greptimedb/issues/7913"

 [scenario]
@@ -48,20 +48,54 @@ semantic = "timestamp"
 [scenario.layout]
 regions = 1
 sst_count = 1024
-rows_per_sst = 4096
-row_group_size = 512
+rows_per_sst = 32768
+row_group_size = 8192
 series_count = 4096
-start_unix_nanos = 1704067200000000000 # 2024-01-01T00:00:00Z
-step_nanos = 1000000000
+start_unix_nanos = 1_704_067_200_000_000_000 # 2024-01-01T00:00:00Z
+step_nanos = 1_000_000_000
 time_range_layout = "non_overlapping_per_sst"
-series_layout = "round_robin"
+series_layout = "timestamp_major"

 [[scenario.queries]]
-name = "narrow_window"
+name = "selector_5m_high_cardinality"
 kind = "tql"
-query = "TQL ANALYZE VERBOSE (1704070800000, 1704070860000, '15s') promql_pushdown_7913{host=~'host.*'}"
-warmup = 2
-iterations = 5
+# TQL start/end timestamps are epoch seconds. `timestamp_major` layout writes one
+# sample for every series at each scrape timestamp, so short PromQL windows still
+# scan millions of raw samples instead of millisecond-scale synthetic trickles.
+# The default 5-minute lookback means scans start before each query window.
+query = "TQL ANALYZE VERBOSE (1704069000, 1704069300, '15s') promql_pushdown_7913{host=~'host.*'}"
+warmup = 3
+iterations = 9
+
+[scenario.queries.thresholds]
+max_candidate_latency_regression_pct = 15
+
+[[scenario.queries]]
+name = "latest_timestamp_equality_subquery"
+kind = "sql"
+query = "SELECT * FROM promql_pushdown_7913 WHERE greptime_timestamp = (SELECT greptime_timestamp FROM promql_pushdown_7913 ORDER BY greptime_timestamp DESC LIMIT 1)"
+warmup = 3
+iterations = 9
+
+[scenario.queries.thresholds]
+max_candidate_latency_regression_pct = 15
+
+[[scenario.queries]]
+name = "selector_15m_high_cardinality"
+kind = "tql"
+query = "TQL ANALYZE VERBOSE (1704069000, 1704069900, '15s') promql_pushdown_7913{host=~'host.*'}"
+warmup = 3
+iterations = 9
+
+[scenario.queries.thresholds]
+max_candidate_latency_regression_pct = 15
+
+[[scenario.queries]]
+name = "selector_1h_high_cardinality"
+kind = "tql"
+query = "TQL ANALYZE VERBOSE (1704069000, 1704072600, '15s') promql_pushdown_7913{host=~'host.*'}"
+warmup = 3
+iterations = 9

 [scenario.queries.thresholds]
 max_candidate_latency_regression_pct = 15
--- a/tests/perf/query_regression_runner.py
+++ b/tests/perf/query_regression_runner.py
@@ -18,6 +18,8 @@
 from __future__ import annotations

 import argparse
+import fcntl
+import hashlib
 import json
 import os
 import re
@@ -36,6 +38,9 @@ from pathlib import Path
 from typing import Any


+FICLONE = 0x40049409
+
+
@dataclass(frozen=True)
 class RunTarget:
    name: str
@@ -61,7 +66,9 @@ def parse_args() -> argparse.Namespace:
    p.add_argument("--base-bin", required=True, type=Path)
    p.add_argument("--candidate-bin", required=True, type=Path)
    p.add_argument("--fixture-generator", type=Path)
+    p.add_argument("--remote-write-generator", type=Path, help="prom_remote_write_fixture helper binary")
    p.add_argument("--work-dir", required=True, type=Path)
+    p.add_argument("--fixture-cache-dir", type=Path, help="persistent directory for generated fixtures, keyed by case content")
    p.add_argument("--reuse-fixture", action="store_true")
    p.add_argument("--allow-large-fixture", action="store_true")
    p.add_argument("--dry-run", action="store_true")
@@ -129,6 +136,10 @@ def sql_ident(name: str) -> str:
    return '"' + name.replace('"', '""') + '"'


+def sql_string(value: str) -> str:
+    return "'" + value.replace("'", "''") + "'"
+
+
 def column_sql(col: dict[str, Any]) -> str:
    return f"{sql_ident(col['name'])} {col['type']}"

@@ -136,15 +147,22 @@ def column_sql(col: dict[str, Any]) -> str:
 def scenario(case: dict[str, Any]) -> dict[str, Any]:
    value = case.get("scenario")
    if not isinstance(value, dict):
-        raise ValueError("case requires [scenario] with kind = 'direct_readable_sst'")
+        raise ValueError("case requires [scenario] with kind = 'direct_readable_sst' or 'prom_remote_write_then_query'")
    kind = value.get("kind")
-    if kind != "direct_readable_sst":
-        raise ValueError(f"unsupported scenario kind {kind!r}; only 'direct_readable_sst' is supported")
+    if kind not in ("direct_readable_sst", "prom_remote_write_then_query"):
+        raise ValueError(f"unsupported scenario kind {kind!r}; supported: 'direct_readable_sst', 'prom_remote_write_then_query'")
    return value


 def case_tables(case: dict[str, Any]) -> list[dict[str, Any]]:
    value = scenario(case)
+    if value.get("kind") == "prom_remote_write_then_query":
+        remote = value.get("remote_write") or {}
+        metric = remote.get("metric") or remote.get("metric_name")
+        database = remote.get("database", "public")
+        if not metric:
+            raise ValueError("remote-write scenario requires scenario.remote_write.metric")
+        return [{"database": database, "name": metric, "engine": "metric", "validate_show_create_engine": False}]
    tables = value.get("tables") or []
    if not tables or value.get("layout", {}).get("regions") != 1:
        raise ValueError("runner supports one or more tables and exactly one region per table")
@@ -163,6 +181,21 @@ def fixture_subdir(table: dict[str, Any], index: int) -> str:
    return safe or f"table_{index:02d}"


+def safe_path_component(raw: str) -> str:
+    safe = re.sub(r"[^A-Za-z0-9_.-]+", "_", raw).strip("._-")
+    return safe or "query_perf_case"
+
+
+def fixture_root(work_root: Path, case_path: Path, case: dict[str, Any], fixture_cache_dir: Path | None) -> Path:
+    if fixture_cache_dir is None:
+        return work_root / "fixture"
+    case_name = case.get("case", {}).get("name") or case_path.parent.name or case_path.stem
+    fixture_config = dict(scenario(case))
+    fixture_config.pop("queries", None)
+    digest = hashlib.sha256(json.dumps(fixture_config, sort_keys=True).encode()).hexdigest()[:16]
+    return fixture_cache_dir.resolve() / f"{safe_path_component(case_name)}-{digest}"
+
+
 def table_fixture_dir(root: Path, tables: list[dict[str, Any]], table: dict[str, Any], index: int) -> Path:
    return root if len(tables) == 1 else root / fixture_subdir(table, index)

@@ -347,15 +380,17 @@ class DistributedCluster:
        )
        wait_health(self.target.datanode_http_port)

-    def start_frontend(self) -> None:
+    def start_frontend(self, config_file: Path | None = None) -> None:
        self._ensure_metasrv_alive()
        log_dir = self.target.work_dir / "logs" / "frontend"
-        self._spawn(
-            "frontend",
-            [
+        cmd = [
                str(self.target.binary),
                "frontend",
                "start",
+        ]
+        if config_file is not None:
+            cmd += ["--config-file", str(config_file)]
+        cmd += [
                "--metasrv-addrs",
                f"127.0.0.1:{self.target.metasrv_rpc_port}",
                "--http-addr",
@@ -370,14 +405,14 @@ class DistributedCluster:
                f"127.0.0.1:{self.target.postgres_port}",
                "--log-dir",
                str(log_dir),
-            ],
-        )
+        ]
+        self._spawn("frontend", cmd)
        wait_health(self.target.http_port)

-    def start_all(self) -> None:
+    def start_all(self, frontend_config: Path | None = None) -> None:
        self.start_metasrv()
        self.start_datanode()
-        self.start_frontend()
+        self.start_frontend(frontend_config)

    def stop_component(self, name: str) -> None:
        proc = self.procs.pop(name, None)
@@ -501,10 +536,15 @@ def generate_fixture(generator: Path | None, case_path: Path, fixture_dir: Path,
    if generator is None:
        return {"status": "skipped", "reason": "no fixture generator provided"}
    summary_path = fixture_dir / "summary.json"
+    reuse_miss: str | None = None
    if reuse_fixture and summary_path.exists():
        summary = json.loads(summary_path.read_text())
-        assert_fixture_summary(summary, region_id=region_id, table_dir=table_dir, region_dir=region_dir, table_name=table_name, database=database)
-        return {"status": "reused", "fixture_dir": str(fixture_dir), "summary": summary}
+        try:
+            assert_fixture_summary(summary, region_id=region_id, table_dir=table_dir, region_dir=region_dir, table_name=table_name, database=database)
+            return {"status": "reused", "fixture_dir": str(fixture_dir), "summary": summary}
+        except RuntimeError as e:
+            reuse_miss = str(e)
+            shutil.rmtree(fixture_dir)
    if fixture_dir.exists() and not reuse_fixture:
        shutil.rmtree(fixture_dir)
    fixture_dir.mkdir(parents=True, exist_ok=True)
@@ -526,19 +566,76 @@ def generate_fixture(generator: Path | None, case_path: Path, fixture_dir: Path,
    summary = json.loads(summary_path.read_text())
    assert_fixture_summary(summary, region_id=region_id, table_dir=table_dir, region_dir=region_dir, table_name=table_name, database=database)
    result["summary"] = summary
+    if reuse_miss is not None:
+        result["reuse_miss"] = reuse_miss
    return result


-def copy_tree_contents(src: Path, dst: Path) -> None:
+def copy_stats() -> dict[str, int]:
+    return {"files": 0, "dirs": 0, "reflinked": 0, "hardlinked": 0, "copied": 0}
+
+
+def merge_copy_stats(left: dict[str, int], right: dict[str, int]) -> None:
+    for key, value in right.items():
+        left[key] = left.get(key, 0) + value
+
+
+def reflink_file(src: Path, dst: Path) -> bool:
+    try:
+        dst.parent.mkdir(parents=True, exist_ok=True)
+        sfd = os.open(src, os.O_RDONLY)
+        try:
+            dfd = os.open(dst, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, src.stat().st_mode & 0o777)
+            try:
+                fcntl.ioctl(dfd, FICLONE, sfd)
+                shutil.copystat(src, dst, follow_symlinks=True)
+                return True
+            finally:
+                os.close(dfd)
+        finally:
+            os.close(sfd)
+    except OSError:
+        try:
+            dst.unlink()
+        except OSError:
+            pass
+        return False
+
+
+def materialize_file(src: Path, dst: Path, *, allow_hardlink: bool) -> str:
+    if reflink_file(src, dst):
+        return "reflinked"
+    if allow_hardlink:
+        try:
+            dst.parent.mkdir(parents=True, exist_ok=True)
+            try:
+                dst.unlink()
+            except FileNotFoundError:
+                pass
+            os.link(src, dst)
+            return "hardlinked"
+        except OSError:
+            pass
+    dst.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copy2(src, dst)
+    return "copied"
+
+
+def copy_tree_contents(src: Path, dst: Path, *, allow_hardlinks: bool = True) -> dict[str, int]:
    if not src.exists():
        raise FileNotFoundError(f"fixture path does not exist: {src}")
+    stats = copy_stats()
    dst.mkdir(parents=True, exist_ok=True)
    for child in src.iterdir():
        target = dst / child.name
        if child.is_dir():
-            shutil.copytree(child, target, dirs_exist_ok=True)
+            stats["dirs"] += 1
+            merge_copy_stats(stats, copy_tree_contents(child, target, allow_hardlinks=allow_hardlinks))
        else:
-            shutil.copy2(child, target)
+            mode = materialize_file(child, target, allow_hardlink=allow_hardlinks)
+            stats["files"] += 1
+            stats[mode] += 1
+    return stats


 def materialize_fixture(target: RunTarget, *, dry_run: bool, preserve_state: bool, expected_region_dir: str | None = None, fixture_dir: Path | None = None, reset_data: bool = True) -> dict[str, Any]:
@@ -564,15 +661,16 @@ def materialize_fixture(target: RunTarget, *, dry_run: bool, preserve_state: boo
    materialize_root.mkdir(parents=True, exist_ok=True)
    if preserve_state and target_region_dir.exists():
        shutil.rmtree(target_region_dir)
+    object_stats: dict[str, int]
    if preserve_state:
        fixture_region_dir = object_store_dir / region_dir
-        copy_tree_contents(fixture_region_dir, target_region_dir)
+        object_stats = copy_tree_contents(fixture_region_dir, target_region_dir, allow_hardlinks=True)
    else:
-        copy_tree_contents(object_store_dir, materialize_root)
+        object_stats = copy_tree_contents(object_store_dir, materialize_root, allow_hardlinks=True)
    if target_manifest_dir.exists():
        shutil.rmtree(target_manifest_dir)
-    copy_tree_contents(manifest_dir, target_manifest_dir)
-    return {"status": "ok", "summary_path": str(summary_path), "data_dir": str(materialize_root), "region_dir": region_dir, "manifest_dir": str(target_manifest_dir), "preserve_state": preserve_state}
+    manifest_stats = copy_tree_contents(manifest_dir, target_manifest_dir, allow_hardlinks=False)
+    return {"status": "ok", "summary_path": str(summary_path), "data_dir": str(materialize_root), "region_dir": region_dir, "manifest_dir": str(target_manifest_dir), "preserve_state": preserve_state, "copy_stats": {"object_store": object_stats, "manifest": manifest_stats}}


 def response_text(body: Any) -> str:
@@ -584,7 +682,7 @@ def validate_show_create(result: dict[str, Any], table: dict[str, Any]) -> list[
    errors = []
    if table["name"].lower() not in text:
        errors.append("SHOW CREATE output does not contain table name")
-    if "engine" not in text or "mito" not in text:
+    if table.get("validate_show_create_engine", True) and ("engine" not in text or "mito" not in text):
        errors.append("SHOW CREATE output does not mention ENGINE=mito")
    if "append_mode" in table and "append_mode" not in text:
        errors.append("SHOW CREATE output does not mention append_mode")
@@ -692,17 +790,240 @@ def require_fresh_work_dirs(targets: list[RunTarget], *, reuse_work_dir: bool, d
            raise RuntimeError(f"target work_dir exists and is non-empty: {target.work_dir}; pass --reuse-work-dir to override")


+def write_frontend_prom_config(target: RunTarget, remote: dict[str, Any]) -> Path:
+    prom = remote.get("prom_store") or {}
+    path = target.work_dir / "frontend-prom-store.toml"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(
+        "[prom_store]\n"
+        "enable = true\n"
+        "with_metric_engine = true\n"
+        f"pending_rows_flush_interval = {json.dumps(str(prom.get('pending_rows_flush_interval', '1s')))}\n"
+        f"max_batch_rows = {int(prom.get('max_batch_rows', 100000))}\n"
+        f"max_concurrent_flushes = {int(prom.get('max_concurrent_flushes', 256))}\n"
+        f"worker_channel_capacity = {int(prom.get('worker_channel_capacity', 65526))}\n"
+        f"max_inflight_requests = {int(prom.get('max_inflight_requests', 3000))}\n"
+    )
+    return path
+
+
+def run_remote_write(generator: Path | None, target: RunTarget, remote: dict[str, Any], *, dry_run: bool) -> dict[str, Any]:
+    if generator is None:
+        if not dry_run:
+            raise ValueError("--remote-write-generator is required for prom_remote_write_then_query unless --dry-run is set")
+        generator = Path("prom_remote_write_fixture")
+    metric = remote.get("metric") or remote.get("metric_name")
+    physical_table = remote.get("physical_table", "greptime_physical_table")
+    cmd = [
+        str(generator),
+        "--endpoint", f"http://127.0.0.1:{target.http_port}/v1/prometheus/write",
+        "--database", remote.get("database", "public"),
+        "--metric", metric,
+        "--physical-table", physical_table,
+        "--series-count", str(int(remote.get("series_count", 8))),
+        "--samples-per-series", str(int(remote.get("samples_per_series", 30))),
+        "--start-unix-millis", str(int(remote.get("start_unix_millis", 1_704_067_200_000))),
+        "--step-millis", str(int(remote.get("step_millis", 15_000))),
+        "--chunk-series-count", str(int(remote.get("chunk_series_count", remote.get("batch_size", 8)))),
+        "--timeout-seconds", str(int(remote.get("timeout_seconds", 60))),
+    ]
+    if dry_run:
+        return {"status": "dry-run", "cmd": cmd}
+    result = run_command(cmd)
+    result["status"] = "ok" if result["returncode"] == 0 else "failed"
+    if result["returncode"] != 0:
+        raise RuntimeError(f"remote-write generator failed for {target.name}: {result['stderr'][:2000]}")
+    try:
+        result["summary"] = json.loads(result["stdout"])
+    except json.JSONDecodeError:
+        result["summary_parse_error"] = result["stdout"]
+    return result
+
+
+def summarize_remote_write_chunks(chunks: list[dict[str, Any]]) -> dict[str, Any]:
+    summary = {"rows": 0, "samples_written": 0, "batches": 0, "elapsed_seconds": 0.0}
+    for chunk in chunks:
+        chunk_summary = chunk.get("summary") or {}
+        summary["rows"] += int(chunk_summary.get("rows", 0))
+        summary["samples_written"] += int(chunk_summary.get("samples_written", chunk_summary.get("rows", 0)))
+        summary["batches"] += int(chunk_summary.get("batches", 0))
+        summary["elapsed_seconds"] += float(chunk_summary.get("elapsed_seconds", chunk.get("elapsed_seconds", 0.0)))
+    return summary
+
+
+def flush_remote_physical_table(target: RunTarget, db: str, physical_table: str, args: argparse.Namespace, *, dry_run: bool, reason: str, chunk_index: int | None = None) -> dict[str, Any]:
+    if dry_run:
+        return {"status": "dry-run", "physical_table": physical_table, "reason": reason, "chunk_index": chunk_index}
+    result = http_post_sql(target.http_port, f"ADMIN FLUSH_TABLE({sql_string(physical_table)})", db, args.http_timeout)
+    result["physical_table"] = physical_table
+    result["reason"] = reason
+    result["chunk_index"] = chunk_index
+    return result
+
+
+def run_remote_write_ingestion(generator: Path | None, target: RunTarget, remote: dict[str, Any], args: argparse.Namespace, *, dry_run: bool) -> tuple[dict[str, Any], list[dict[str, Any]]]:
+    sample_chunk_size = remote.get("sample_chunk_size")
+    db = remote.get("database", "public")
+    physical_table = remote.get("physical_table", "greptime_physical_table")
+    if sample_chunk_size is None:
+        rw = run_remote_write(generator, target, remote, dry_run=dry_run)
+        flush = flush_remote_physical_table(target, db, physical_table, args, dry_run=dry_run, reason="final")
+        return rw, [flush]
+
+    total_samples = int(remote.get("samples_per_series", 30))
+    chunk_samples = int(sample_chunk_size)
+    if chunk_samples <= 0:
+        raise ValueError("scenario.remote_write.sample_chunk_size must be positive")
+    flush_every = int(remote.get("flush_every_sample_chunks", 1))
+    if flush_every <= 0:
+        raise ValueError("scenario.remote_write.flush_every_sample_chunks must be positive")
+    start = int(remote.get("start_unix_millis", 1_704_067_200_000))
+    step = int(remote.get("step_millis", 15_000))
+    chunks: list[dict[str, Any]] = []
+    flushes: list[dict[str, Any]] = []
+    chunk_index = 0
+    last_flushed_chunk = 0
+    for offset in range(0, total_samples, chunk_samples):
+        current = min(chunk_samples, total_samples - offset)
+        chunk_index += 1
+        chunk_remote = dict(remote)
+        chunk_remote["samples_per_series"] = current
+        chunk_remote["start_unix_millis"] = start + offset * step
+        result = run_remote_write(generator, target, chunk_remote, dry_run=dry_run)
+        result["sample_offset"] = offset
+        result["samples_per_series"] = current
+        result["chunk_index"] = chunk_index
+        chunks.append(result)
+        if chunk_index % flush_every == 0:
+            flushes.append(flush_remote_physical_table(target, db, physical_table, args, dry_run=dry_run, reason="periodic", chunk_index=chunk_index))
+            last_flushed_chunk = chunk_index
+    if last_flushed_chunk != chunk_index:
+        flushes.append(flush_remote_physical_table(target, db, physical_table, args, dry_run=dry_run, reason="final", chunk_index=chunk_index))
+    aggregate = summarize_remote_write_chunks(chunks)
+    if dry_run:
+        series_count = int(remote.get("series_count", 8))
+        chunk_series_count = int(remote.get("chunk_series_count", remote.get("batch_size", 8)))
+        aggregate["rows"] = series_count * total_samples
+        aggregate["samples_written"] = aggregate["rows"]
+        aggregate["batches"] = chunk_index * ((series_count + chunk_series_count - 1) // chunk_series_count)
+    return {
+        "status": "dry-run" if dry_run else ("ok" if all(chunk.get("status") == "ok" for chunk in chunks) else "failed"),
+        "mode": "sample-chunked",
+        "sample_chunk_size": chunk_samples,
+        "flush_every_sample_chunks": flush_every,
+        "chunks": chunks,
+        "aggregate": aggregate,
+    }, flushes
+
+
+def expected_remote_write_rows(remote: dict[str, Any]) -> int:
+    return int(remote.get("series_count", 8)) * int(remote.get("samples_per_series", 30))
+
+
+def extract_count_value(result: dict[str, Any]) -> int | None:
+    body = result.get("response")
+    if not isinstance(body, dict):
+        return None
+    data = body.get("data")
+    if not isinstance(data, list) or not data:
+        return None
+    row = data[0]
+    if not isinstance(row, dict):
+        return None
+    for key, value in row.items():
+        if key.lower() == "count(*)" or key.lower().startswith("count("):
+            try:
+                return int(value)
+            except (TypeError, ValueError):
+                return None
+    return None
+
+
+def poll_expected_count(target: RunTarget, table_name: str, db: str, expected_rows: int, timeout_s: float, http_timeout: float) -> dict[str, Any]:
+    sql = f"SELECT count(*) FROM {sql_ident(table_name)}"
+    deadline = time.monotonic() + timeout_s
+    attempts = 0
+    last: dict[str, Any] | None = None
+    while True:
+        attempts += 1
+        result = http_post_sql(target.http_port, sql, db, http_timeout)
+        observed_rows = extract_count_value(result)
+        result["expected_rows"] = expected_rows
+        result["observed_rows"] = observed_rows
+        result["attempts"] = attempts
+        result["row_count_ok"] = result.get("ok") and observed_rows == expected_rows
+        if result["row_count_ok"]:
+            return result
+        last = result
+        if time.monotonic() >= deadline:
+            break
+        time.sleep(0.5)
+    assert last is not None
+    last["ok"] = False
+    last["row_count_ok"] = False
+    last["error"] = f"expected {expected_rows} rows but observed {last.get('observed_rows')} after {attempts} attempts"
+    return last
+
+
+def run_remote_write_scenario(args: argparse.Namespace, case: dict[str, Any], case_path: Path, targets: list[RunTarget], report: dict[str, Any]) -> None:
+    remote = scenario(case).get("remote_write") or {}
+    tables = case_tables(case)
+    if args.fixture_only:
+        raise ValueError("--fixture-only is not supported for prom_remote_write_then_query; use --dry-run for planning")
+    if args.remote_write_generator is not None:
+        require_binary(args.remote_write_generator, "remote-write generator", dry_run=args.dry_run)
+    elif not args.dry_run:
+        raise ValueError("--remote-write-generator is required for prom_remote_write_then_query")
+    clusters: list[DistributedCluster] = []
+    target_results = []
+    try:
+        for target in targets:
+            target.work_dir.mkdir(parents=True, exist_ok=True)
+            config_path = write_frontend_prom_config(target, remote)
+            cluster = DistributedCluster(target)
+            clusters.append(cluster)
+            if not args.dry_run:
+                cluster.start_all(config_path)
+            db = remote.get("database", "public")
+            create_database = {"status": "dry-run", "database": db}
+            if not args.dry_run:
+                create_database = http_post_sql(target.http_port, f"CREATE DATABASE IF NOT EXISTS {sql_ident(db)}", "public", args.http_timeout)
+            rw, flushes = run_remote_write_ingestion(args.remote_write_generator, target, remote, args, dry_run=args.dry_run)
+            physical = remote.get("physical_table", "greptime_physical_table")
+            visibility = {"status": "dry-run"}
+            query_result = {"validation": [], "validation_errors": [], "measurements": [], "status": "planned"}
+            if not args.dry_run:
+                visibility = poll_expected_count(target, tables[0]["name"], db, expected_remote_write_rows(remote), float(remote.get("visibility_timeout_seconds", 30)), args.http_timeout)
+                query_result = run_queries(target, case, tables, args.http_timeout)
+            tr = {"name": target.name, "binary": str(target.binary), "work_dir": str(target.work_dir), "components": cluster.component_report(), "frontend_config": str(config_path), "create_database": create_database, "remote_write": rw, "flushes": flushes, "flush": flushes[-1] if flushes else None, "visibility": visibility, **query_result}
+            flushes_ok = all(flush.get("ok") for flush in flushes)
+            remote_checks_ok = args.dry_run or (create_database.get("ok") and flushes_ok and visibility.get("ok") and visibility.get("row_count_ok"))
+            if not remote_checks_ok:
+                tr.setdefault("validation_errors", []).append({"phase": "remote_write_visibility", "create_database_ok": create_database.get("ok"), "flushes_ok": flushes_ok, "visibility_ok": visibility.get("ok"), "row_count_ok": visibility.get("row_count_ok"), "create_database": create_database, "flushes": flushes, "visibility": visibility})
+            tr["status"] = "planned" if args.dry_run else ("measured" if remote_checks_ok and query_result["status"] == "ok" else "failed")
+            write_json(target.report_path, tr)
+            report["targets"].append(tr)
+            target_results.append(query_result)
+        if not args.dry_run:
+            report["thresholds"] = enforce_thresholds(case, target_results[0], target_results[1])
+        report["status"] = "planned" if args.dry_run else ("failed" if any(t["status"] == "failed" for t in report["thresholds"]) or any(t.get("status") == "failed" for t in report["targets"]) else "ok")
+    finally:
+        for cluster in reversed(clusters):
+            cluster.stop_all()
+
+
 def main() -> int:
    args = parse_args()
    case_path = args.case.resolve()
    case = load_case(case_path)
    scenario_config = scenario(case)
+    scenario_kind = scenario_config.get("kind")
    tables = case_tables(case)
    work_root = args.work_dir.resolve()
    work_root.mkdir(parents=True, exist_ok=True)
    require_binary(args.base_bin, "base", dry_run=args.dry_run or args.fixture_only)
    require_binary(args.candidate_bin, "candidate", dry_run=args.dry_run or args.fixture_only)
-    if not args.dry_run and args.fixture_generator is None:
+    if scenario_kind == "direct_readable_sst" and not args.dry_run and args.fixture_generator is None:
        raise ValueError("--fixture-generator is required unless --dry-run is set")
    if args.fixture_generator is not None:
        require_binary(args.fixture_generator, "fixture generator", dry_run=args.dry_run)
@@ -710,14 +1031,25 @@ def main() -> int:
    ports = allocate_ports(16)
    targets = [make_target("base", args.base_bin.resolve(), work_root, ports[:8]), make_target("candidate", args.candidate_bin.resolve(), work_root, ports[8:])]
    require_fresh_work_dirs(targets, reuse_work_dir=args.reuse_work_dir, dry_run=args.dry_run, fixture_only=args.fixture_only)
-    fixture_dir = work_root / "fixture"
-    report: dict[str, Any] = {"case_path": str(case_path), "case": case.get("case", {}), "scenario": scenario_config, "queries": planned_queries(case), "dry_run": args.dry_run, "fixture_only": args.fixture_only, "query_mode": "fixture-only" if args.fixture_only else "distributed", "reuse_work_dir": args.reuse_work_dir, "http_timeout": args.http_timeout, "targets": [], "thresholds": [], "status": "planned" if args.dry_run else "running"}
+    fixture_dir = fixture_root(work_root, case_path, case, args.fixture_cache_dir)
+    reuse_fixture = args.reuse_fixture or args.fixture_cache_dir is not None
+    report: dict[str, Any] = {"case_path": str(case_path), "case": case.get("case", {}), "scenario": scenario_config, "queries": planned_queries(case), "dry_run": args.dry_run, "fixture_only": args.fixture_only, "query_mode": "fixture-only" if args.fixture_only else "distributed", "reuse_work_dir": args.reuse_work_dir, "reuse_fixture": reuse_fixture, "fixture_cache_dir": str(args.fixture_cache_dir.resolve()) if args.fixture_cache_dir is not None else None, "fixture_dir": str(fixture_dir), "http_timeout": args.http_timeout, "targets": [], "thresholds": [], "status": "planned" if args.dry_run else "running"}
+
+    if scenario_kind == "prom_remote_write_then_query":
+        try:
+            run_remote_write_scenario(args, case, case_path, targets, report)
+        except Exception as e:  # noqa: BLE001 - write machine-readable failure report
+            report["status"] = "failed"
+            report["error"] = repr(e)
+        write_json(work_root / "query-regression-report.json", report)
+        print(json.dumps(report, indent=2, sort_keys=True))
+        return 1 if report["status"] == "failed" else 0

    if args.fixture_only or args.dry_run:
        generations = []
        for table_idx, table in enumerate(tables):
            per_table_fixture_dir = table_fixture_dir(fixture_dir, tables, table, table_idx)
-            generations.append(generate_fixture(args.fixture_generator, case_path, per_table_fixture_dir, dry_run=args.dry_run, reuse_fixture=args.reuse_fixture, allow_large_fixture=args.allow_large_fixture, table_name=table["name"] if len(tables) > 1 else None, database=table["database"] if len(tables) > 1 else None))
+            generations.append(generate_fixture(args.fixture_generator, case_path, per_table_fixture_dir, dry_run=args.dry_run, reuse_fixture=reuse_fixture, allow_large_fixture=args.allow_large_fixture, table_name=table["name"] if len(tables) > 1 else None, database=table["database"] if len(tables) > 1 else None))
        report["fixture_generation"] = generations[0] if len(generations) == 1 else generations
        for target in targets:
            target.work_dir.mkdir(parents=True, exist_ok=True)
@@ -759,7 +1091,7 @@ def main() -> int:
        generations = []
        for table_idx, meta in enumerate(discovered[0]):
            per_table_fixture_dir = table_fixture_dir(fixture_dir, tables, tables[table_idx], table_idx)
-            generations.append(generate_fixture(args.fixture_generator, case_path, per_table_fixture_dir, dry_run=False, reuse_fixture=args.reuse_fixture, allow_large_fixture=args.allow_large_fixture, table_name=meta["table"] if len(tables) > 1 else None, database=meta["schema"] if len(tables) > 1 else None, region_id=meta["region_id"], table_dir=meta["table_dir"], region_dir=meta["region_dir"]))
+            generations.append(generate_fixture(args.fixture_generator, case_path, per_table_fixture_dir, dry_run=False, reuse_fixture=reuse_fixture, allow_large_fixture=args.allow_large_fixture, table_name=meta["table"] if len(tables) > 1 else None, database=meta["schema"] if len(tables) > 1 else None, region_id=meta["region_id"], table_dir=meta["table_dir"], region_dir=meta["region_dir"]))
        report["fixture_generation"] = generations[0] if len(generations) == 1 else generations

        target_results = []