From efc23d79bd30a5132cbbeeb82b135e339fe72f91 Mon Sep 17 00:00:00 2001
From: discord9 <55937128+discord9@users.noreply.github.com>
Date: Thu, 6 Aug 2026 14:19:52 +0800
Subject: [PATCH] feat(perf): add mixed read/write saturated case for scheduler
 validation

Add optional [scenario.write_measure.mix] to write_throughput: runs the
remote-write ingestion in a background thread while query-loop threads
(query_parallelism x query_interval_ms) hammer the same frontend, so
query and write tasks contend on the datanode runtime under a dual
backlog. Adds query gates (failure rate, p99 regression) and a
best-effort scheduler poll-share diagnostic scraped from the datanode
/metrics. Includes write_read_mixed_scheduler built-in case.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
---
 src/cmd/src/bin/query_perf_fixture/case.rs    | 260 ++++++++++++++++
 tests/perf/README.md                          |  60 ++++
 .../write_read_mixed_scheduler/case.toml      |  76 +++++
 tests/perf/query_regression_runner.py         | 289 +++++++++++++++++-
 tests/perf/test_write_throughput_runner.py    | 248 +++++++++++++++
 5 files changed, 929 insertions(+), 4 deletions(-)
 create mode 100644 tests/perf/query_cases/write_read_mixed_scheduler/case.toml

diff --git a/src/cmd/src/bin/query_perf_fixture/case.rs b/src/cmd/src/bin/query_perf_fixture/case.rs
index 8474b52d0a..86c1345081 100644
--- a/src/cmd/src/bin/query_perf_fixture/case.rs
+++ b/src/cmd/src/bin/query_perf_fixture/case.rs
@@ -153,6 +153,85 @@ pub(super) struct WriteMeasureConfig {
     #[serde(default)]
     pub(super) target_rps: f64,
     pub(super) thresholds: WriteThroughputThresholds,
+    /// Optional concurrent read+write ("mix") measurement. When present the
+    /// runner runs the remote-write ingestion in a background thread while a
+    /// query loop hammers the same frontend for `duration_seconds`, so query
+    /// and write tasks genuinely contend on the datanode runtime under a dual
+    /// backlog.
+    #[serde(default)]
+    pub(super) mix: Option<MixMeasureConfig>,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub(super) struct MixMeasureConfig {
+    /// SQL for the concurrent query loop. Defaults to a `count(*)` over the
+    /// remote-write physical table (`greptime_physical_table`), which scans
+    /// every ingested row and is therefore a real datanode-runtime contender.
+    #[serde(default)]
+    pub(super) query_sql: Option<String>,
+    /// Wall-clock interval between queries per loop thread, in milliseconds.
+    #[serde(default = "default_mix_query_interval_ms")]
+    pub(super) query_interval_ms: NonZeroU64,
+    /// Number of concurrent query-loop threads.
+    #[serde(default = "default_mix_query_parallelism")]
+    pub(super) query_parallelism: NonZeroU64,
+    pub(super) thresholds: MixMeasureThresholds,
+}
+
+pub(super) fn default_mix_query_interval_ms() -> NonZeroU64 {
+    NonZeroU64::new(100).expect("100 is non-zero")
+}
+pub(super) fn default_mix_query_parallelism() -> NonZeroU64 {
+    NonZeroU64::new(1).expect("1 is non-zero")
+}
+
+impl MixMeasureConfig {
+    pub(super) fn validate(&self) -> Result<(), String> {
+        if let Some(sql) = &self.query_sql
+            && sql.trim().is_empty()
+        {
+            return Err(
+                "scenario.write_measure.mix.query_sql must be a non-empty SQL string".to_string(),
+            );
+        }
+        self.thresholds.validate()
+    }
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+pub(super) struct MixMeasureThresholds {
+    /// Max fraction of failed query attempts per target, in [0, 1] (0.05 = 5%).
+    pub(super) max_query_failure_rate: f64,
+    /// Max candidate-vs-base query p99 latency regression percent:
+    /// `(candidate_p99_ms - base_p99_ms) / base_p99_ms * 100`.
+    pub(super) max_query_p99_regression_pct: f64,
+}
+
+impl MixMeasureThresholds {
+    fn validate(&self) -> Result<(), String> {
+        for (name, value) in [
+            ("max_query_failure_rate", self.max_query_failure_rate),
+            (
+                "max_query_p99_regression_pct",
+                self.max_query_p99_regression_pct,
+            ),
+        ] {
+            if !value.is_finite() || value < 0.0 {
+                return Err(format!(
+                    "scenario.write_measure.mix.thresholds.{name} must be a finite non-negative number"
+                ));
+            }
+        }
+        if self.max_query_failure_rate > 1.0 {
+            return Err(
+                "scenario.write_measure.mix.thresholds.max_query_failure_rate must be <= 1.0 (a rate in [0, 1])"
+                    .to_string(),
+            );
+        }
+        Ok(())
+    }
 }
 
 #[derive(Debug, Deserialize, Serialize)]
@@ -185,6 +264,9 @@ impl WriteMeasureConfig {
                     .to_string(),
             );
         }
+        if let Some(mix) = &self.mix {
+            mix.validate()?;
+        }
         self.thresholds.validate()
     }
 }
@@ -670,6 +752,10 @@ mod tests {
         repo_root().join("tests/perf/query_cases/write_throughput_scheduler/case.toml")
     }
 
+    fn builtin_write_read_mixed_case() -> PathBuf {
+        repo_root().join("tests/perf/query_cases/write_read_mixed_scheduler/case.toml")
+    }
+
     fn parse_case(text: &str) -> Result<CaseFile, toml::de::Error> {
         toml::from_str(text)
     }
@@ -719,6 +805,31 @@ max_p99_latency_regression_pct = 10.0
 max_concurrent_polls = 16
 query_weight = 2
 write_weight = 8
+"#;
+
+    const MIXED_CASE: &str = r#"
+[scenario]
+kind = "write_throughput"
+
+[scenario.remote_write]
+metric = "write_read_mixed_test"
+
+[scenario.write_measure]
+duration_seconds = 60
+window_seconds = 5
+
+[scenario.write_measure.thresholds]
+max_failure_rate = 0.05
+max_mean_rps_regression_pct = 10.0
+max_p99_latency_regression_pct = 10.0
+
+[scenario.write_measure.mix]
+query_interval_ms = 100
+query_parallelism = 2
+
+[scenario.write_measure.mix.thresholds]
+max_query_failure_rate = 0.05
+max_query_p99_regression_pct = 10.0
 "#;
 
     #[test]
@@ -884,4 +995,153 @@ write_weight = 8
             .expect_err("max_failure_rate above 1.0 must be rejected");
         assert!(err.contains("max_failure_rate"), "{err}");
     }
+
+    #[test]
+    fn parses_builtin_write_read_mixed_case() {
+        let text = std::fs::read_to_string(builtin_write_read_mixed_case())
+            .expect("built-in write_read_mixed case.toml must exist");
+        let case = parse_case(&text).expect("built-in write_read_mixed case must parse");
+        assert_eq!(case.scenario.kind(), "write_throughput");
+        let scenario = write_throughput_scenario(&text);
+        assert_eq!(scenario.remote_write.series_count, 2048);
+        assert_eq!(scenario.remote_write.samples_per_series, 3600);
+        assert_eq!(scenario.write_measure.duration_seconds.get(), 60);
+        assert_eq!(scenario.write_measure.window_seconds.get(), 5);
+        let mix = scenario
+            .write_measure
+            .mix
+            .as_ref()
+            .expect("built-in mixed case must declare a mix section");
+        assert_eq!(mix.query_interval_ms.get(), 100);
+        assert_eq!(mix.query_parallelism.get(), 2);
+        assert_eq!(mix.thresholds.max_query_failure_rate, 0.05);
+        assert_eq!(mix.thresholds.max_query_p99_regression_pct, 10.0);
+        let scheduler = scenario
+            .scheduler
+            .as_ref()
+            .expect("built-in mixed case must declare a scheduler section");
+        assert_eq!(scheduler.query_weight, 2);
+        assert_eq!(scheduler.write_weight, 8);
+        scenario
+            .write_measure
+            .validate()
+            .expect("built-in mixed case must validate");
+        scheduler
+            .validate()
+            .expect("built-in scheduler must validate");
+    }
+
+    #[test]
+    fn write_throughput_mix_uses_defaults_and_validates() {
+        // Defaults apply only for keys the case does not set explicitly; drop
+        // the explicit query_parallelism to exercise the default.
+        let defaults_case = MIXED_CASE.replace("query_parallelism = 2\n", "");
+        let scenario = write_throughput_scenario(&defaults_case);
+        let mix = scenario
+            .write_measure
+            .mix
+            .as_ref()
+            .expect("mix section must parse into Some");
+        assert!(mix.query_sql.is_none());
+        assert_eq!(mix.query_interval_ms.get(), 100);
+        assert_eq!(mix.query_parallelism.get(), 1);
+        mix.validate().expect("mix config must validate");
+        scenario
+            .write_measure
+            .validate()
+            .expect("write_measure with mix must validate");
+    }
+
+    #[test]
+    fn write_throughput_mix_parses_explicit_values() {
+        let explicit = MIXED_CASE.replace(
+            "query_interval_ms = 100\nquery_parallelism = 2",
+            "query_sql = \"SELECT count(*) FROM greptime_physical_table\"\nquery_interval_ms = 250\nquery_parallelism = 4",
+        );
+        let scenario = write_throughput_scenario(&explicit);
+        let mix = scenario
+            .write_measure
+            .mix
+            .expect("mix section must parse into Some");
+        assert_eq!(
+            mix.query_sql.as_deref(),
+            Some("SELECT count(*) FROM greptime_physical_table")
+        );
+        assert_eq!(mix.query_interval_ms.get(), 250);
+        assert_eq!(mix.query_parallelism.get(), 4);
+        mix.validate().expect("explicit mix config must validate");
+    }
+
+    #[test]
+    fn write_throughput_mix_rejects_zero_interval_or_parallelism() {
+        let zero_interval = MIXED_CASE.replace("query_interval_ms = 100", "query_interval_ms = 0");
+        assert!(
+            parse_case(&zero_interval).is_err(),
+            "zero query_interval_ms must be rejected"
+        );
+
+        let zero_parallelism = MIXED_CASE.replace("query_parallelism = 2", "query_parallelism = 0");
+        assert!(
+            parse_case(&zero_parallelism).is_err(),
+            "zero query_parallelism must be rejected"
+        );
+    }
+
+    #[test]
+    fn write_throughput_mix_rejects_unknown_fields() {
+        let unknown = MIXED_CASE.replace(
+            "query_interval_ms = 100",
+            "query_interval_ms = 100\nbogus_mix_field = 1",
+        );
+        let err = parse_case(&unknown).expect_err("unknown mix field must be rejected");
+        assert!(err.to_string().contains("bogus_mix_field"), "{err}");
+    }
+
+    #[test]
+    fn write_throughput_mix_rejects_empty_query_sql() {
+        let empty_sql = MIXED_CASE.replace(
+            "query_interval_ms = 100",
+            "query_sql = \"   \"\nquery_interval_ms = 100",
+        );
+        let scenario = write_throughput_scenario(&empty_sql);
+        let mix = scenario
+            .write_measure
+            .mix
+            .expect("mix section must parse into Some");
+        let err = mix
+            .validate()
+            .expect_err("blank query_sql must be rejected");
+        assert!(err.contains("query_sql"), "{err}");
+    }
+
+    #[test]
+    fn write_throughput_mix_rejects_invalid_thresholds() {
+        let negative_limit = MIXED_CASE.replace(
+            "max_query_p99_regression_pct = 10.0",
+            "max_query_p99_regression_pct = -1.0",
+        );
+        let scenario = write_throughput_scenario(&negative_limit);
+        let mix = scenario
+            .write_measure
+            .mix
+            .expect("mix section must parse into Some");
+        let err = mix
+            .validate()
+            .expect_err("negative query p99 regression limit must be rejected");
+        assert!(err.contains("max_query_p99_regression_pct"), "{err}");
+
+        let oversized_rate = MIXED_CASE.replace(
+            "max_query_failure_rate = 0.05",
+            "max_query_failure_rate = 1.5",
+        );
+        let scenario = write_throughput_scenario(&oversized_rate);
+        let mix = scenario
+            .write_measure
+            .mix
+            .expect("mix section must parse into Some");
+        let err = mix
+            .validate()
+            .expect_err("query failure rate above 1.0 must be rejected");
+        assert!(err.contains("max_query_failure_rate"), "{err}");
+    }
 }
diff --git a/tests/perf/README.md b/tests/perf/README.md
index 373316cafc..9a291dd504 100644
--- a/tests/perf/README.md
+++ b/tests/perf/README.md
@@ -386,6 +386,66 @@ for a max-throughput run. For local results, run the case at least three times
 on an otherwise idle machine and compare median regressions rather than relying
 on one run.
 
+### Mixed read/write (saturated) measurement
+
+Adding `[scenario.write_measure.mix]` turns the pure-write scenario into a
+concurrent read+write ("mix") measurement: the runner spawns the chunked
+remote-write ingestion in a background thread (each chunk is still its own
+`query_perf_fixture prom-remote-write` subprocess) and, while it runs, drives
+`query_parallelism` query-loop threads against the same frontend HTTP
+`/v1/sql` endpoint for `duration_seconds`, then joins the ingestion. Both
+streams hit the same frontend/datanode, so query tasks and write tasks
+genuinely contend on the datanode runtime under a dual backlog — the
+workload-scheduler 2:8 allocation can be observed end to end.
+
+```toml
+[scenario.write_measure.mix]
+# Default query is `SELECT count(*) FROM greptime_physical_table` (the
+# physical table where every remote-write row lands); override with query_sql.
+# query_interval_ms is the per-thread wall-clock interval between queries.
+query_interval_ms = 100
+query_parallelism = 2
+
+[scenario.write_measure.mix.thresholds]
+max_query_failure_rate = 0.05        # per-target failed query attempt fraction
+max_query_p99_regression_pct = 10.0  # (candidate_p99 - base_p99) / base_p99 * 100
+```
+
+Rust owns the mix schema: `query_interval_ms` (default 100) and
+`query_parallelism` (default 1) are positive integers (zero is rejected at
+parse time), `query_sql` when present must be non-empty, and the mix
+thresholds follow the same finite/non-negative rules as the write thresholds
+with `max_query_failure_rate` in `[0, 1]`. Unknown mix fields are rejected by
+`query_perf_fixture plan`.
+
+For each target the mixed run records the existing `write_measurement` plus:
+
+- `query_measurement`: `{samples, failures, failure_rate, p50_ms, p99_ms,
+  mean_ms, latency_samples}`. Latency percentiles/mean cover every attempt
+  that recorded a latency, including failed ones, so timeouts show up in p99.
+- `scheduler_poll_deltas`: best-effort deltas of the datanode cumulative
+  `greptime_workload_scheduler_polls{workload="query"|"write"}` counters over
+  the window, scraped from `http://127.0.0.1:<datanode_http_port>/metrics`.
+  This is a diagnostic only, never a gate: the base target runs with the
+  scheduler disabled and typically exposes no such samples, and a workload
+  whose counter is missing or reset (decreased) is reported as `null`.
+
+Threshold enforcement combines the existing write gates with the query gates:
+per-target `max_query_failure_rate` and base-vs-candidate
+`max_query_p99_regression_pct` (positive `actual_pct` is a candidate
+regression, negative is an improvement). The case gates that neither side
+collapses — it does not gate on the 80% write poll share, which is the
+scheduler micro-benchmark's job.
+
+`tests/perf/query_cases/write_read_mixed_scheduler/case.toml` is the built-in
+mixed case: the same 2048 series × 3600 samples remote-write sizing as
+`write_throughput_scheduler`, a 60s measurement window, mix query
+`SELECT count(*) FROM greptime_physical_table` at 100ms interval with
+parallelism 2, and gates of ≤ 5% write/query failure, ≤ 10% write RPS
+regression, ≤ 10% write and query p99 regression, and a 15k rows/s absolute
+floor. It is opt-in (not part of the default CI case set); run it the same way
+as the pure write-throughput case, substituting the case path.
+
 ## Generator contract
 
 The direct-SST generator should accept a case definition with:
diff --git a/tests/perf/query_cases/write_read_mixed_scheduler/case.toml b/tests/perf/query_cases/write_read_mixed_scheduler/case.toml
new file mode 100644
index 0000000000..aa4577e939
--- /dev/null
+++ b/tests/perf/query_cases/write_read_mixed_scheduler/case.toml
@@ -0,0 +1,76 @@
+# Mixed read/write (saturated) regression case for the workload scheduler
+# (PR #8736).
+#
+# Runs remote-write ingestion in the background while a query loop hammers the
+# same frontend concurrently for the whole measurement window, so query tasks
+# and write tasks genuinely contend on the datanode runtime under a dual
+# backlog. With the candidate scheduler's 2:8 query:write weights, admitted
+# polls should skew ~80% write while neither side collapses; this case does
+# NOT gate on the poll-share number (that is the scheduler micro-benchmark's
+# job) but gates that neither side regresses: the existing write gates
+# (failure, mean-RPS regression, p99 regression, absolute RPS floor) plus the
+# query gates (failure rate, p99 regression).
+#
+# Sized for a local run: 2048 series x 3600 samples = 7,372,800 rows per
+# target, same as write_throughput_scheduler. The ingestion is split into 20
+# sample chunks of 180 samples per series; the measurement spans the first 60s
+# (12 windows of 5s), during which the query loop (parallelism 2, one query
+# every 100ms per thread) runs concurrently against the same frontend.
+
+[case]
+name = "write_read_mixed_scheduler"
+description = "Concurrent saturated read+write throughput/latency for validating workload-scheduler 2:8 allocation under dual backlog"
+issue = "https://github.com/GreptimeTeam/greptimedb/pull/8736"
+
+[scenario]
+kind = "write_throughput"
+
+[scenario.scheduler]
+# enable is not set here: the runner derives it per target (base = disabled,
+# candidate = enabled). max_concurrent_polls 0 = 4 * global_rt_size.
+max_concurrent_polls = 16
+query_weight = 2
+write_weight = 8
+
+[scenario.remote_write]
+database = "public"
+metric = "write_read_mixed_scheduler"
+physical_table = "greptime_physical_table"
+series_count = 2048
+samples_per_series = 3600
+sample_chunk_size = 180
+flush_every_sample_chunks = 1
+start_unix_millis = 1_704_067_200_000 # 2024-01-01T00:00:00Z
+step_millis = 1000
+chunk_series_count = 256
+timeout_seconds = 120
+
+[scenario.remote_write.prom_store]
+pending_rows_flush_interval = "1s"
+max_batch_rows = 1_000_000
+max_concurrent_flushes = 256
+
+[scenario.write_measure]
+duration_seconds = 60
+window_seconds = 5
+target_rps = 0 # 0 = max throughput; the achieved rate is measured
+
+# Concurrent read+write measurement: the runner spawns the remote-write
+# ingestion in a background thread and runs `query_parallelism` query-loop
+# threads (one query every `query_interval_ms` each) against the same frontend
+# for `duration_seconds`. The default query is
+# `SELECT count(*) FROM greptime_physical_table` (the physical table where all
+# remote-write rows land); override with `query_sql`.
+[scenario.write_measure.mix]
+query_interval_ms = 100
+query_parallelism = 2
+
+[scenario.write_measure.mix.thresholds]
+max_query_failure_rate = 0.05        # per-target failed query attempt fraction
+max_query_p99_regression_pct = 10.0  # (candidate_p99 - base_p99) / base_p99 * 100
+
+[scenario.write_measure.thresholds]
+max_failure_rate = 0.05
+max_mean_rps_regression_pct = 10
+max_p99_latency_regression_pct = 10
+min_rps_absolute = 15000
diff --git a/tests/perf/query_regression_runner.py b/tests/perf/query_regression_runner.py
index a282862bfa..e2615fc3cc 100644
--- a/tests/perf/query_regression_runner.py
+++ b/tests/perf/query_regression_runner.py
@@ -29,6 +29,7 @@ import socket
 import statistics
 import subprocess
 import sys
+import threading
 import time
 import tomllib
 import urllib.error
@@ -47,6 +48,9 @@ OTLP_TRACE_METRICS = {
     "greptime_servers_http_otlp_traces_elapsed_count",
 }
 PROMETHEUS_SAMPLE_RE = re.compile(r"^([A-Za-z_:][A-Za-z0-9_:]*)(?:\{.*\})?\s+([^\s]+)(?:\s+.*)?$")
+# Workload-scheduler cumulative poll counters on the datanode /metrics
+# endpoint, e.g. greptime_workload_scheduler_polls{workload="write"} 1234.
+SCHEDULER_POLL_RE = re.compile(r'^greptime_workload_scheduler_polls\{[^}]*workload="(query|write)"[^}]*\}\s+(\d+)(?:\s+.*)?$')
 
 
 @dataclass(frozen=True)
@@ -1125,6 +1129,55 @@ def metric_delta(after: dict[str, Any], before: dict[str, Any], name: str) -> fl
     return delta
 
 
+def parse_scheduler_poll_metrics(text: str) -> dict[str, int]:
+    """Parse `greptime_workload_scheduler_polls{workload="..."}` samples.
+
+    Returns {workload: cumulative polls} for the workloads the datanode
+    scheduler tracks ("query", "write"); workloads absent from the scrape are
+    omitted (the base target runs with the scheduler disabled and typically
+    exposes no such samples at all).
+    """
+    values: dict[str, int] = {}
+    for line in text.splitlines():
+        match = SCHEDULER_POLL_RE.match(line.strip())
+        if match:
+            values[match.group(1)] = int(match.group(2))
+    return values
+
+
+def scrape_scheduler_polls(target: RunTarget, http_timeout: float) -> dict[str, Any]:
+    """Best-effort datanode scheduler-poll snapshot (never a gate).
+
+    Returns {"captured_monotonic_seconds", "values": {workload: polls}} on
+    success or {"error", "values": {}} when the datanode /metrics endpoint is
+    unreachable or the scheduler metric is absent.
+    """
+    try:
+        with urllib.request.urlopen(f"http://127.0.0.1:{target.datanode_http_port}/metrics", timeout=http_timeout) as response:
+            text = response.read().decode()
+        return {"captured_monotonic_seconds": time.monotonic(), "values": parse_scheduler_poll_metrics(text)}
+    except Exception as e:  # noqa: BLE001 - best-effort diagnostic
+        return {"error": repr(e), "values": {}}
+
+
+def scheduler_poll_deltas(after: dict[str, Any], before: dict[str, Any]) -> dict[str, Any]:
+    """Delta of cumulative scheduler polls between two snapshots, per workload.
+
+    A workload whose counter is missing in either snapshot, or that decreased
+    (counter reset/restart), is reported as None because its delta is unknown.
+    """
+    result: dict[str, Any] = {}
+    for workload in ("query", "write"):
+        before_value = before.get("values", {}).get(workload)
+        after_value = after.get("values", {}).get(workload)
+        if before_value is None or after_value is None:
+            result[workload] = None
+        else:
+            delta = after_value - before_value
+            result[workload] = delta if delta >= 0 else None
+    return result
+
+
 def otelgen_command(otelgen_bin: Path, target: RunTarget, load: dict[str, Any]) -> list[str]:
     return [
         str(otelgen_bin),
@@ -1667,6 +1720,202 @@ def enforce_write_throughput_thresholds(write_measure: dict[str, Any], base: dic
     return results
 
 
+def mix_query_sql(mix: dict[str, Any], remote: dict[str, Any]) -> str:
+    """Resolve the mixed-scenario query SQL.
+
+    Defaults to a count(*) over the remote-write physical table, which scans
+    every ingested row and therefore contends with the write path on the
+    datanode runtime.
+    """
+    sql = mix.get("query_sql")
+    if sql:
+        return str(sql)
+    return f"SELECT count(*) FROM {sql_ident(remote['physical_table'])}"
+
+
+def expected_mix_query_attempts(duration_seconds: int, interval_ms: int, parallelism: int) -> int:
+    """Per-thread attempt count for a query loop: queries at t=0, interval,
+    2*interval, ... while t < duration. Wall-clock jitter may drop at most one
+    trailing query per thread, so the observed count is
+    [expected - parallelism, expected]."""
+    per_thread = math.ceil(max(0.0, float(duration_seconds)) * 1000.0 / max(1, int(interval_ms)))
+    return per_thread * max(1, int(parallelism))
+
+
+def run_mix_query_loop(target: RunTarget, mix: dict[str, Any], db: str, duration_seconds: int, http_timeout: float, remote: dict[str, Any]) -> list[dict[str, Any]]:
+    """Run `query_parallelism` query-loop threads for `duration_seconds`.
+
+    Each thread issues the mix query via HTTP SQL every `query_interval_ms`
+    (per-thread wall clock; if a query itself takes longer than the interval
+    the thread does not sleep). Returns the collected attempt results in
+    completion order; every attempt carries `ok`, `status`, `latency_ms`, and
+    `sql` from `http_post_sql`.
+    """
+    query_sql = mix_query_sql(mix, remote)
+    interval_ms = int(mix.get("query_interval_ms", 100))
+    parallelism = max(1, int(mix.get("query_parallelism", 1)))
+    attempts: list[dict[str, Any]] = []
+    lock = threading.Lock()
+    stop = threading.Event()
+    deadline = time.monotonic() + max(0.0, float(duration_seconds))
+    threads = []
+    for _ in range(parallelism):
+        thread = threading.Thread(
+            target=_mix_query_worker,
+            args=(target, query_sql, db, interval_ms, deadline, http_timeout, attempts, lock, stop),
+            daemon=True,
+        )
+        threads.append(thread)
+        thread.start()
+    for thread in threads:
+        # Bounded join: workers may be stuck in a slow HTTP call beyond the
+        # deadline; their daemon status keeps process exit unblocked.
+        thread.join(timeout=max(10.0, float(interval_ms) / 1000.0 + 2.0))
+    return attempts
+
+
+def _mix_query_worker(
+    target: RunTarget,
+    query_sql: str,
+    db: str,
+    interval_ms: int,
+    deadline: float,
+    http_timeout: float,
+    attempts: list[dict[str, Any]],
+    lock: threading.Lock,
+    stop: threading.Event,
+) -> None:
+    interval_seconds = float(interval_ms) / 1000.0
+    while not stop.is_set():
+        started = time.monotonic()
+        if started >= deadline:
+            return
+        result = http_post_sql(target.http_port, query_sql, db, http_timeout)
+        with lock:
+            attempts.append(result)
+        remaining = interval_seconds - (time.monotonic() - started)
+        if remaining > 0 and stop.wait(remaining):
+            return
+
+
+def mix_query_measurement(attempts: list[dict[str, Any]]) -> dict[str, Any]:
+    """Aggregate query-loop attempts into {samples, failures, failure_rate,
+    p50_ms, p99_ms, mean_ms, latency_samples}.
+
+    Latency percentiles/mean cover every attempt that recorded a latency
+    (including failed ones, so timeouts show up in p99). `latency_samples`
+    counts those attempts. The input is snapshotted so a still-running daemon
+    query worker cannot mutate it mid-aggregation.
+    """
+    attempts = list(attempts)
+    total = len(attempts)
+    failures = sum(1 for a in attempts if not a.get("ok"))
+    latencies = [float(a["latency_ms"]) for a in attempts if a.get("latency_ms") is not None]
+    return {
+        "samples": total,
+        "failures": failures,
+        "failure_rate": failures / total if total else None,
+        "p50_ms": percentile(latencies, 50) if latencies else None,
+        "p99_ms": percentile(latencies, 99) if latencies else None,
+        "mean_ms": statistics.mean(latencies) if latencies else None,
+        "latency_samples": len(latencies),
+    }
+
+
+def planned_mix_query_measurement(mix: dict[str, Any], write_measure: dict[str, Any]) -> dict[str, Any]:
+    """Dry-run query measurement: no query loop ran, so report planned counts."""
+    duration_seconds = int(write_measure["duration_seconds"])
+    interval_ms = int(mix.get("query_interval_ms", 100))
+    parallelism = int(mix.get("query_parallelism", 1))
+    return {
+        "status": "planned",
+        "planned_attempts": expected_mix_query_attempts(duration_seconds, interval_ms, parallelism),
+        "query_interval_ms": interval_ms,
+        "query_parallelism": parallelism,
+        "samples": 0,
+        "failures": 0,
+        "failure_rate": None,
+        "p50_ms": None,
+        "p99_ms": None,
+        "mean_ms": None,
+        "latency_samples": 0,
+    }
+
+
+def run_mixed_ingestion_and_queries(generator: Path | None, target: RunTarget, remote: dict[str, Any], args: argparse.Namespace, mix: dict[str, Any], write_measure: dict[str, Any], *, dry_run: bool) -> tuple[dict[str, Any], list[dict[str, Any]], list[dict[str, Any]]]:
+    """Run remote-write ingestion in a background thread while a query loop
+    hammers the same frontend concurrently for `duration_seconds`, then join.
+
+    The ingestion is the same chunked `run_write_throughput_ingestion`
+    machinery, just executed from a thread; each chunk is its own
+    `query_perf_fixture prom-remote-write` subprocess (generate+send in one
+    shot), and the query loop issues HTTP /v1/sql requests from the main
+    process, so both hit the same frontend/datanode and genuinely contend on
+    the datanode runtime. Returns (rw, flushes, query_attempts); in dry-run
+    mode no thread and no query loop is started and attempts is empty.
+    """
+    if dry_run:
+        rw, flushes = run_write_throughput_ingestion(generator, target, remote, args, dry_run=True)
+        return rw, flushes, []
+
+    results: dict[str, Any] = {}
+    errors: list[BaseException] = []
+
+    def ingest() -> None:
+        try:
+            results["rw"], results["flushes"] = run_write_throughput_ingestion(generator, target, remote, args, dry_run=False)
+        except BaseException as e:  # noqa: BLE001 - re-raised in the caller
+            errors.append(e)
+
+    thread = threading.Thread(target=ingest, name=f"{target.name}-ingest", daemon=True)
+    thread.start()
+    try:
+        attempts = run_mix_query_loop(target, mix, remote["database"], int(write_measure["duration_seconds"]), args.http_timeout, remote)
+    finally:
+        thread.join()
+        if thread.is_alive():
+            # The datanode may be wedged; the ingestion thread is daemon so it
+            # cannot block process exit, but surface the lack of results.
+            raise RuntimeError(f"write ingestion thread did not finish for {target.name}")
+    if errors:
+        raise errors[0]
+    return results["rw"], results["flushes"], attempts
+
+
+def planned_mix_query_thresholds(mix: dict[str, Any]) -> list[dict[str, Any]]:
+    thresholds = mix["thresholds"]
+    return [
+        {"threshold": "max_query_failure_rate", "status": "planned", "limit": thresholds["max_query_failure_rate"]},
+        {"threshold": "max_query_p99_regression_pct", "status": "planned", "limit_pct": thresholds["max_query_p99_regression_pct"]},
+    ]
+
+
+def enforce_mix_query_thresholds(mix: dict[str, Any], base: dict[str, Any], candidate: dict[str, Any]) -> list[dict[str, Any]]:
+    """Enforce the query-side gates of the mixed read/write scenario.
+
+    Per-target: `max_query_failure_rate` (failed query attempts / total).
+    Base-vs-candidate: `max_query_p99_regression_pct`
+    `(candidate_p99_ms - base_p99_ms) / base_p99_ms * 100`; positive actual
+    means a candidate regression, negative is an improvement.
+    """
+    thresholds = mix["thresholds"]
+    results: list[dict[str, Any]] = []
+    failure_limit = float(thresholds["max_query_failure_rate"])
+    for target_name, measurement in (("base", base), ("candidate", candidate)):
+        failure_rate = measurement.get("failure_rate")
+        results.append({"target": target_name, "threshold": "max_query_failure_rate", "status": "passed" if failure_rate is not None and failure_rate <= failure_limit else "failed", "actual": failure_rate, "limit": failure_limit})
+
+    base_p99 = base.get("p99_ms")
+    candidate_p99 = candidate.get("p99_ms")
+    p99_limit = float(thresholds["max_query_p99_regression_pct"])
+    if base_p99 in (None, 0) or candidate_p99 is None:
+        results.append({"threshold": "max_query_p99_regression_pct", "status": "failed", "reason": "missing or zero query p99 latency", "base": base_p99, "candidate": candidate_p99})
+    else:
+        actual = (candidate_p99 - base_p99) / base_p99 * 100.0
+        results.append({"threshold": "max_query_p99_regression_pct", "status": "passed" if actual <= p99_limit else "failed", "actual_pct": actual, "limit_pct": p99_limit, "base": base_p99, "candidate": candidate_p99})
+    return results
+
+
 def extract_count_value(result: dict[str, Any]) -> int | None:
     body = result.get("response")
     if not isinstance(body, dict):
@@ -1790,6 +2039,7 @@ def run_write_throughput_scenario(args: argparse.Namespace, case: dict[str, Any]
 
     clusters: list[DistributedCluster] = []
     measurements: list[dict[str, Any]] = []
+    query_measurements: list[dict[str, Any]] = []
     try:
         for target in targets:
             target.work_dir.mkdir(parents=True, exist_ok=True)
@@ -1802,7 +2052,19 @@ def run_write_throughput_scenario(args: argparse.Namespace, case: dict[str, Any]
             create_database: dict[str, Any] = {"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_write_throughput_ingestion(helper, target, remote, args, dry_run=args.dry_run)
+            mix = write_measure.get("mix")
+            scheduler_polls_before: dict[str, Any] | None = None
+            scheduler_polls_after: dict[str, Any] | None = None
+            if mix is not None and not args.dry_run:
+                scheduler_polls_before = scrape_scheduler_polls(target, args.http_timeout)
+            if mix is not None:
+                rw, flushes, query_attempts = run_mixed_ingestion_and_queries(helper, target, remote, args, mix, write_measure, dry_run=args.dry_run)
+                query_measurement = planned_mix_query_measurement(mix, write_measure) if args.dry_run else mix_query_measurement(query_attempts)
+                if not args.dry_run:
+                    scheduler_polls_after = scrape_scheduler_polls(target, args.http_timeout)
+            else:
+                rw, flushes = run_write_throughput_ingestion(helper, target, remote, args, dry_run=args.dry_run)
+                query_measurement = None
             measurement = planned_write_throughput_measurement(remote, write_measure) if args.dry_run else write_throughput_measurement(rw, write_measure)
             tr = {
                 "name": target.name,
@@ -1817,17 +2079,36 @@ def run_write_throughput_scenario(args: argparse.Namespace, case: dict[str, Any]
                 "scheduler": scheduler_report_entry(target.name, scenario_config.get("scheduler")),
                 "write_measurement": measurement,
             }
+            if query_measurement is not None:
+                tr["query_measurement"] = query_measurement
+            if mix is not None:
+                tr["mix"] = mix
+                tr["scheduler_poll_deltas"] = scheduler_poll_deltas(scheduler_polls_after, scheduler_polls_before) if scheduler_polls_after is not None else {"status": "planned"}
             flushes_ok = all(flush.get("ok") for flush in flushes)
             measurement_ok = measurement.get("mean_rps") not in (None, 0)
-            checks_ok = args.dry_run or (create_database.get("ok") and flushes_ok and measurement_ok)
+            query_ok = query_measurement is None or query_measurement.get("samples", 0) > 0
+            checks_ok = args.dry_run or (create_database.get("ok") and flushes_ok and measurement_ok and query_ok)
             if not checks_ok and not args.dry_run:
-                tr.setdefault("validation_errors", []).append({"phase": "write_throughput", "create_database_ok": create_database.get("ok"), "flushes_ok": flushes_ok, "measurement_ok": measurement_ok, "mean_rps": measurement.get("mean_rps"), "failure_rate": measurement.get("failure_rate")})
+                validation = {"phase": "write_throughput", "create_database_ok": create_database.get("ok"), "flushes_ok": flushes_ok, "measurement_ok": measurement_ok, "mean_rps": measurement.get("mean_rps"), "failure_rate": measurement.get("failure_rate")}
+                if query_measurement is not None:
+                    validation["query_ok"] = query_ok
+                    validation["query_samples"] = query_measurement.get("samples")
+                tr.setdefault("validation_errors", []).append(validation)
             tr["status"] = "planned" if args.dry_run else ("measured" if checks_ok else "failed")
             write_json(target.report_path, tr)
             report["targets"].append(tr)
             measurements.append(measurement)
+            if query_measurement is not None:
+                query_measurements.append(query_measurement)
             cluster.stop_all()
-        report["thresholds"] = planned_write_throughput_thresholds(write_measure) if args.dry_run else enforce_write_throughput_thresholds(write_measure, measurements[0], measurements[1])
+        if args.dry_run:
+            report["thresholds"] = planned_write_throughput_thresholds(write_measure)
+            if write_measure.get("mix") is not None:
+                report["thresholds"] += planned_mix_query_thresholds(write_measure["mix"])
+        else:
+            report["thresholds"] = enforce_write_throughput_thresholds(write_measure, measurements[0], measurements[1])
+            if write_measure.get("mix") is not None:
+                report["thresholds"] += enforce_mix_query_thresholds(write_measure["mix"], query_measurements[0], query_measurements[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):
diff --git a/tests/perf/test_write_throughput_runner.py b/tests/perf/test_write_throughput_runner.py
index 1d7cee2ceb..121eb45e73 100644
--- a/tests/perf/test_write_throughput_runner.py
+++ b/tests/perf/test_write_throughput_runner.py
@@ -23,6 +23,7 @@ dry-run path. No live database, cluster, or generator subprocess is started.
 import importlib.util
 import sys
 import tempfile
+import time
 import unittest
 from pathlib import Path
 from unittest.mock import patch
@@ -79,6 +80,19 @@ def make_write_measure(**overrides):
     return write_measure
 
 
+def make_mix(**overrides):
+    mix = {
+        "query_interval_ms": 100,
+        "query_parallelism": 2,
+        "thresholds": {
+            "max_query_failure_rate": 0.05,
+            "max_query_p99_regression_pct": 10.0,
+        },
+    }
+    mix.update(overrides)
+    return mix
+
+
 class WriteThroughputDispatchTest(unittest.TestCase):
     def test_scenario_accepts_write_throughput_kind(self) -> None:
         case = {"scenario": {"kind": "write_throughput", "remote_write": {}, "write_measure": {}}}
@@ -406,5 +420,239 @@ class WriteThroughputScenarioDryRunTest(unittest.TestCase):
             self.assertTrue(targets[0].report_path.exists())
 
 
+class WriteThroughputMixQueryTest(unittest.TestCase):
+    """Pure-function coverage for the mixed read/write query loop."""
+
+    def test_expected_attempts_math(self) -> None:
+        self.assertEqual(runner.expected_mix_query_attempts(60, 100, 2), 1200)
+        self.assertEqual(runner.expected_mix_query_attempts(10, 3000, 1), 4)
+        self.assertEqual(runner.expected_mix_query_attempts(60, 1000, 4), 240)
+        # Degenerate parallelism is clamped to 1 (interval is already
+        # guaranteed > 0 by the Rust schema; the clamp is defensive).
+        self.assertEqual(runner.expected_mix_query_attempts(60, 100, 0), 600)
+
+    def test_mix_query_sql_default_and_override(self) -> None:
+        remote = make_remote(physical_table="greptime_physical_table")
+        mix = make_mix()
+        self.assertEqual(runner.mix_query_sql(mix, remote), 'SELECT count(*) FROM "greptime_physical_table"')
+        mix["query_sql"] = "SELECT max(greptime_value) FROM greptime_physical_table"
+        self.assertEqual(runner.mix_query_sql(mix, remote), "SELECT max(greptime_value) FROM greptime_physical_table")
+
+    def test_query_loop_runs_and_collects_attempts(self) -> None:
+        target = runner.RunTarget("base", Path("/bin/true"), Path("/tmp/wt"), Path("/tmp/wt/data"), Path("/tmp/wt/fixture"), Path("/tmp/wt/report.json"), 4000, 4001, 4002, 4003, 4004, 4005, 4006, 4007, Path("/tmp/wt/datanode/data"))
+        mix = make_mix(query_interval_ms=100, query_parallelism=2)
+
+        def fake_http(port, sql, db, timeout):
+            return {"ok": True, "status": 200, "latency_ms": 5.0, "response": {"data": []}, "sql": sql}
+
+        with patch.object(runner, "http_post_sql", side_effect=fake_http):
+            attempts = runner.run_mix_query_loop(target, mix, "public", duration_seconds=1, http_timeout=5.0, remote=make_remote())
+        # 1s at 100ms per thread = 10 attempts/thread; a delayed thread may
+        # drop its final query, but both threads must have run repeatedly.
+        self.assertGreaterEqual(len(attempts), 12)
+        self.assertLessEqual(len(attempts), 20)
+        self.assertTrue(all(a["ok"] for a in attempts))
+        self.assertTrue(all(a["sql"] == 'SELECT count(*) FROM "greptime_physical_table"' for a in attempts))
+
+    def test_mixed_ingestion_and_queries_join_thread(self) -> None:
+        target = runner.RunTarget("base", Path("/bin/true"), Path("/tmp/wt"), Path("/tmp/wt/data"), Path("/tmp/wt/fixture"), Path("/tmp/wt/report.json"), 4000, 4001, 4002, 4003, 4004, 4005, 4006, 4007, Path("/tmp/wt/datanode/data"))
+        args = runner.argparse.Namespace(http_timeout=5.0)
+        remote = make_remote()
+
+        def fake_ingest(generator, target, remote, args, *, dry_run):
+            time.sleep(0.05)
+            return {"status": "ok", "chunks": []}, [{"ok": True}]
+
+        with (
+            patch.object(runner, "run_write_throughput_ingestion", side_effect=fake_ingest),
+            patch.object(runner, "run_mix_query_loop", return_value=[{"ok": True, "latency_ms": 1.0}]),
+        ):
+            rw, flushes, attempts = runner.run_mixed_ingestion_and_queries(None, target, remote, args, make_mix(), make_write_measure(), dry_run=False)
+        self.assertEqual(rw["status"], "ok")
+        self.assertEqual(flushes, [{"ok": True}])
+        self.assertEqual(attempts, [{"ok": True, "latency_ms": 1.0}])
+
+    def test_mixed_dry_run_skips_threads(self) -> None:
+        target = runner.RunTarget("base", Path("/bin/true"), Path("/tmp/wt"), Path("/tmp/wt/data"), Path("/tmp/wt/fixture"), Path("/tmp/wt/report.json"), 4000, 4001, 4002, 4003, 4004, 4005, 4006, 4007, Path("/tmp/wt/datanode/data"))
+        args = runner.argparse.Namespace(http_timeout=5.0)
+        with (
+            patch.object(runner, "run_write_throughput_ingestion", return_value=({"status": "ok"}, [{"ok": True}])) as ingest,
+            patch.object(runner.threading, "Thread") as thread_cls,
+        ):
+            rw, flushes, attempts = runner.run_mixed_ingestion_and_queries(None, target, make_remote(), args, make_mix(), make_write_measure(), dry_run=True)
+        thread_cls.assert_not_called()
+        ingest.assert_called_once()
+        self.assertEqual(attempts, [])
+
+
+class WriteThroughputMixMeasurementTest(unittest.TestCase):
+    def test_measurement_aggregates_latency_and_failures(self) -> None:
+        attempts = [
+            {"ok": True, "latency_ms": 10.0},
+            {"ok": True, "latency_ms": 20.0},
+            {"ok": True, "latency_ms": 30.0},
+            {"ok": False, "latency_ms": 40.0},
+            {"ok": False, "latency_ms": 50.0},
+        ]
+        measurement = runner.mix_query_measurement(attempts)
+        self.assertEqual(measurement["samples"], 5)
+        self.assertEqual(measurement["failures"], 2)
+        self.assertEqual(measurement["failure_rate"], 0.4)
+        self.assertEqual(measurement["latency_samples"], 5)
+        self.assertEqual(measurement["p50_ms"], 30.0)
+        self.assertEqual(measurement["p99_ms"], 50.0)
+        self.assertAlmostEqual(measurement["mean_ms"], 30.0)
+
+    def test_measurement_empty_attempts(self) -> None:
+        measurement = runner.mix_query_measurement([])
+        self.assertEqual(measurement["samples"], 0)
+        self.assertEqual(measurement["failures"], 0)
+        self.assertIsNone(measurement["failure_rate"])
+        self.assertIsNone(measurement["p50_ms"])
+        self.assertIsNone(measurement["p99_ms"])
+        self.assertIsNone(measurement["mean_ms"])
+
+    def test_planned_measurement_reports_expected_attempts(self) -> None:
+        measurement = runner.planned_mix_query_measurement(make_mix(), make_write_measure())
+        self.assertEqual(measurement["status"], "planned")
+        self.assertEqual(measurement["planned_attempts"], 1200)
+        self.assertEqual(measurement["query_interval_ms"], 100)
+        self.assertEqual(measurement["query_parallelism"], 2)
+        self.assertIsNone(measurement["p99_ms"])
+
+
+class WriteThroughputMixThresholdTest(unittest.TestCase):
+    def test_combined_write_and_query_gates(self) -> None:
+        write_base = {"mean_rps": 100_000, "p99_latency_ms": 50.0, "failure_rate": 0.0}
+        write_candidate = {"mean_rps": 90_000, "p99_latency_ms": 55.0, "failure_rate": 0.02}
+        query_base = {"p99_ms": 200.0, "failure_rate": 0.0}
+        query_candidate = {"p99_ms": 220.0, "failure_rate": 0.01}
+        write_results = runner.enforce_write_throughput_thresholds(make_write_measure(), write_base, write_candidate)
+        query_results = runner.enforce_mix_query_thresholds(make_mix(), query_base, query_candidate)
+        combined = write_results + query_results
+        by_key = {(r.get("target"), r["threshold"]): r for r in combined}
+        self.assertEqual(len(combined), 9)
+        # Write gates unchanged.
+        self.assertEqual(by_key[(None, "max_mean_rps_regression_pct")]["status"], "passed")
+        self.assertEqual(by_key[(None, "max_p99_latency_regression_pct")]["status"], "passed")
+        # Query gates pass within limits.
+        self.assertEqual(by_key[("base", "max_query_failure_rate")]["status"], "passed")
+        self.assertEqual(by_key[("candidate", "max_query_failure_rate")]["status"], "passed")
+        q99 = by_key[(None, "max_query_p99_regression_pct")]
+        self.assertEqual(q99["status"], "passed")
+        self.assertAlmostEqual(q99["actual_pct"], 10.0)
+        self.assertAlmostEqual(q99["limit_pct"], 10.0)
+
+    def test_query_gates_fail_on_regression_and_failures(self) -> None:
+        query_base = {"p99_ms": 200.0, "failure_rate": 0.0}
+        query_candidate = {"p99_ms": 260.0, "failure_rate": 0.10}
+        results = runner.enforce_mix_query_thresholds(make_mix(), query_base, query_candidate)
+        by_key = {(r.get("target"), r["threshold"]): r for r in results}
+        self.assertEqual(by_key[("candidate", "max_query_failure_rate")]["status"], "failed")
+        self.assertEqual(by_key[(None, "max_query_p99_regression_pct")]["status"], "failed")
+        self.assertAlmostEqual(by_key[(None, "max_query_p99_regression_pct")]["actual_pct"], 30.0)
+
+    def test_query_gates_fail_on_missing_base(self) -> None:
+        query_base = {"p99_ms": None, "failure_rate": None}
+        query_candidate = {"p99_ms": 260.0, "failure_rate": 0.0}
+        results = runner.enforce_mix_query_thresholds(make_mix(), query_base, query_candidate)
+        by_key = {(r.get("target"), r["threshold"]): r for r in results}
+        self.assertEqual(by_key[("base", "max_query_failure_rate")]["status"], "failed")
+        self.assertEqual(by_key[(None, "max_query_p99_regression_pct")]["status"], "failed")
+        self.assertEqual(by_key[(None, "max_query_p99_regression_pct")]["reason"], "missing or zero query p99 latency")
+
+    def test_planned_mix_query_thresholds(self) -> None:
+        planned = runner.planned_mix_query_thresholds(make_mix())
+        self.assertEqual([p["threshold"] for p in planned], ["max_query_failure_rate", "max_query_p99_regression_pct"])
+        self.assertTrue(all(p["status"] == "planned" for p in planned))
+
+
+class SchedulerPollMetricsTest(unittest.TestCase):
+    def test_parse_scheduler_poll_metrics(self) -> None:
+        text = (
+            "# HELP greptime_workload_scheduler_polls Cumulative task polls admitted by the workload scheduler\n"
+            "# TYPE greptime_workload_scheduler_polls gauge\n"
+            "greptime_workload_scheduler_polls{workload=\"query\"} 1234\n"
+            "greptime_workload_scheduler_polls{workload=\"write\"} 5678\n"
+            "greptime_workload_scheduler_queued_tasks{workload=\"query\"} 3\n"
+            "greptime_runtime_threads_alive{thread_name=\"global\"} 8\n"
+        )
+        self.assertEqual(runner.parse_scheduler_poll_metrics(text), {"query": 1234, "write": 5678})
+        # Scheduler disabled: no samples at all.
+        self.assertEqual(runner.parse_scheduler_poll_metrics("greptime_runtime_threads_alive 8\n"), {})
+
+    def test_scheduler_poll_deltas(self) -> None:
+        before = {"values": {"query": 100, "write": 900}}
+        after = {"values": {"query": 180, "write": 1780}}
+        self.assertEqual(runner.scheduler_poll_deltas(after, before), {"query": 80, "write": 880})
+
+    def test_scheduler_poll_deltas_unknown_on_missing_or_reset(self) -> None:
+        before = {"values": {"query": 100}}
+        after = {"values": {"query": 180, "write": 20}}
+        # write absent before -> unknown; write decreased -> unknown (reset).
+        self.assertEqual(runner.scheduler_poll_deltas(after, before), {"query": 80, "write": None})
+        before = {"values": {"query": 100, "write": 900}}
+        after = {"values": {}}
+        self.assertEqual(runner.scheduler_poll_deltas(after, before), {"query": None, "write": None})
+
+
+class WriteThroughputMixScenarioDryRunTest(unittest.TestCase):
+    def test_dry_run_with_mix_plans_write_and_query_gates(self) -> None:
+        events = []
+
+        class FakeCluster:
+            def __init__(self, target):
+                self.target = target
+                self.stopped = False
+                events.append(f"create:{self.target.name}")
+
+            def component_report(self):
+                return {}
+
+            def stop_all(self):
+                if not self.stopped:
+                    self.stopped = True
+                    events.append(f"stop:{self.target.name}")
+
+        args = runner.argparse.Namespace(
+            fixture_only=False,
+            fixture_generator=Path("query_perf_fixture"),
+            remote_write_generator=None,
+            dry_run=True,
+            http_timeout=1.0,
+        )
+        write_measure = make_write_measure()
+        write_measure["mix"] = make_mix()
+        case = {"scenario": {"kind": "write_throughput", "remote_write": make_remote(), "write_measure": write_measure}}
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            root = Path(tmpdir)
+            targets = [
+                runner.make_target("base", Path("/bin/true"), root, list(range(10_000, 10_008))),
+                runner.make_target("candidate", Path("/bin/true"), root, list(range(10_008, 10_016))),
+            ]
+            report = {"targets": []}
+            with (
+                patch.object(runner, "DistributedCluster", FakeCluster),
+                patch.object(runner, "run_command") as run_cmd,
+            ):
+                runner.run_write_throughput_scenario(args, case, Path("case.toml"), targets, report)
+
+            run_cmd.assert_not_called()
+            self.assertEqual(report["status"], "planned")
+            self.assertEqual(len(report["thresholds"]), 6)
+            self.assertTrue(all(t["status"] == "planned" for t in report["thresholds"]))
+            self.assertEqual(
+                [t["threshold"] for t in report["thresholds"]],
+                ["max_failure_rate", "max_mean_rps_regression_pct", "max_p99_latency_regression_pct", "min_rps_absolute", "max_query_failure_rate", "max_query_p99_regression_pct"],
+            )
+            for tr in report["targets"]:
+                self.assertEqual(tr["write_measurement"]["status"], "planned")
+                self.assertEqual(tr["query_measurement"]["status"], "planned")
+                self.assertEqual(tr["query_measurement"]["planned_attempts"], 1200)
+                self.assertEqual(tr["scheduler_poll_deltas"], {"status": "planned"})
+                self.assertEqual(tr["mix"]["query_parallelism"], 2)
+
+
 if __name__ == "__main__":
     unittest.main()