test(perf): exercise positive inverted appliers without result caches

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
(cherry picked from commit b8286499d0)
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

.github/scripts/query-regression-run.py

@@ -130,6 +130,7 @@ class RunTarget:
     datanode_http_port: int
     datanode_data_dir: Path
     frontend_config: Path | None = None
+    datanode_config: Path | None = None
 
 
 def allocate_ports(n: int) -> list[int]:
@@ -151,6 +152,7 @@ def make_target(
     root: Path,
     ports: list[int],
     frontend_config: Path | None = None,
+    datanode_config: Path | None = None,
 ) -> RunTarget:
     work_dir = root / name
     return RunTarget(
@@ -167,6 +169,7 @@ def make_target(
         datanode_http_port=ports[3],
         datanode_data_dir=work_dir / "datanode-0" / "data",
         frontend_config=frontend_config,
+        datanode_config=datanode_config,
     )
 
 
@@ -200,7 +203,7 @@ def component_command(target: RunTarget, name: str) -> list[str]:
             "--log-dir", str(log_dir),
         ]
     if name == "datanode":
-        return [
+        command = [
             str(target.binary), "datanode", "start",
             "--grpc-bind-addr", f"127.0.0.1:{target.datanode_rpc_port}",
             "--grpc-server-addr", f"127.0.0.1:{target.datanode_rpc_port}",
@@ -208,6 +211,9 @@ def component_command(target: RunTarget, name: str) -> list[str]:
             "--data-home", str(target.datanode_data_dir), "--log-dir", str(log_dir),
             "--node-id", "0", "--metasrv-addrs", f"127.0.0.1:{target.metasrv_rpc_port}",
         ]
+        if target.datanode_config is not None:
+            command.extend(["--config-file", str(target.datanode_config)])
+        return command
     if name == "frontend":
         command = [str(target.binary), "frontend", "start"]
         if target.frontend_config is not None:
@@ -407,6 +413,7 @@ def run_remote_case(
             work_dir,
             ports[:8],
             work_dir / "base" / "frontend-prom-store.toml",
+            work_dir / "base" / "datanode-index-cache.toml",
         ),
         make_target(
             "candidate",
@@ -414,6 +421,7 @@ def run_remote_case(
             work_dir,
             ports[8:],
             work_dir / "candidate" / "frontend-prom-store.toml",
+            work_dir / "candidate" / "datanode-index-cache.toml",
         ),
     ]
     for target in targets:
@@ -430,6 +438,16 @@ def run_remote_case(
         render_status = subprocess.run(render, check=False).returncode
         if render_status != 0:
             return render_status
+        if target.datanode_config is None:
+            raise RuntimeError(f"remote target has no datanode config path: {target.name}")
+        render_datanode = [
+            str(runner), "render-remote-config", "--case", str(case_path),
+            "--fixture-generator", str(fixture_generator),
+            "--output", str(target.datanode_config), "--target", "datanode",
+        ]
+        render_status = subprocess.run(render_datanode, check=False).returncode
+        if render_status != 0:
+            return render_status
 
     procs: dict[tuple[str, str], subprocess.Popen[bytes]] = {}
     report = work_dir / "query-regression-report.json"

src/cmd/src/bin/query_perf_fixture/case.rs

@@ -116,6 +116,8 @@ pub(super) struct PromRemoteWritePlan {
     pub(super) storage: Option<StorageConfig>,
     #[serde(default)]
     pub(super) read_bench: Option<ReadBenchConfig>,
+    #[serde(default)]
+    pub(super) disable_index_result_cache: bool,
 }
 
 pub(super) fn default_database() -> String {
@@ -462,6 +464,7 @@ metric = "metric"
         };
         assert!(scenario.remote_write.base_setup_sql.is_empty());
         assert!(scenario.remote_write.candidate_setup_sql.is_empty());
+        assert!(!scenario.remote_write.disable_index_result_cache);
     }
 
     #[test]
@@ -473,6 +476,7 @@ kind = "prom_remote_write_then_query"
 
 [scenario.remote_write]
 metric = "metric"
+disable_index_result_cache = true
 base_setup_sql = ["CREATE TABLE base_table"]
 candidate_setup_sql = ["CREATE TABLE candidate_table", "ALTER TABLE candidate_table SET 'x'='y'"]
 "#,
@@ -484,6 +488,7 @@ candidate_setup_sql = ["CREATE TABLE candidate_table", "ALTER TABLE candidate_ta
         let Scenario::PromRemoteWriteThenQuery(scenario) = roundtrip.scenario else {
             panic!("expected prom_remote_write_then_query scenario");
         };
+        assert!(scenario.remote_write.disable_index_result_cache);
         assert_eq!(
             scenario.remote_write.base_setup_sql,
             ["CREATE TABLE base_table"]

src/cmd/src/bin/query_regression_runner/mod.rs

@@ -99,6 +99,14 @@ struct RenderRemoteConfigArgs {
     fixture_generator: PathBuf,
     #[arg(long, value_name = "PATH")]
     output: PathBuf,
+    #[arg(long, value_enum, default_value = "frontend")]
+    target: RemoteConfigTarget,
+}
+
+#[derive(Clone, Copy, Debug, clap::ValueEnum)]
+enum RemoteConfigTarget {
+    Frontend,
+    Datanode,
 }
 
 #[derive(Debug, Parser)]

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

@@ -124,6 +124,8 @@ pub(super) struct RemoteWrite {
     pub(super) value: RemoteValue,
     pub(super) storage: Option<StorageConfig>,
     pub(super) read_bench: Option<ReadBenchConfig>,
+    #[serde(default)]
+    pub(super) disable_index_result_cache: bool,
 }
 
 #[derive(Clone, Debug, Deserialize)]

src/cmd/src/bin/query_regression_runner/remote/ingest.rs

@@ -25,11 +25,21 @@ use crate::query_regression_runner::plan::normalized_remote_write;
 use crate::query_regression_runner::sql::{
     extract_count_value, http_post_sql, sql_ident, sql_string, value_f64, value_u64,
 };
-use crate::query_regression_runner::{PrepareRemoteArgs, RenderRemoteConfigArgs, Result};
+use crate::query_regression_runner::{
+    PrepareRemoteArgs, RemoteConfigTarget, RenderRemoteConfigArgs, Result,
+};
 
 pub(super) async fn run_render_remote_config(args: RenderRemoteConfigArgs) -> Result<()> {
     let (_, remote) = normalized_remote_write(&args.fixture_generator, &args.case)?;
-    fs::write(args.output, frontend_prom_config(&remote.prom_store)?)?;
+    let config = match args.target {
+        RemoteConfigTarget::Frontend => frontend_prom_config(&remote.prom_store)?,
+        RemoteConfigTarget::Datanode if remote.disable_index_result_cache => {
+            "[[region_engine]]\n[region_engine.mito]\nrange_result_cache_size = \"0B\"\n[region_engine.mito.index]\nresult_cache_size = \"0B\"\n"
+                .to_string()
+        }
+        RemoteConfigTarget::Datanode => String::new(),
+    };
+    fs::write(args.output, config)?;
     Ok(())
 }
 
@@ -580,6 +590,7 @@ mod tests {
             sample_chunk_size: Some(2),
             flush_every_sample_chunks: 2,
             visibility_timeout_seconds: 30,
+            disable_index_result_cache: false,
             base_setup_sql: Vec::new(),
             candidate_setup_sql: Vec::new(),
             prom_store: PromStore {

tests/perf/README.md

@@ -193,6 +193,8 @@ before trusting the query measurements. Use
 `--fixture-generator /path/to/query_perf_fixture` to provide the Rust helper to
 the outer driver.
 
+Remote-write cases can set `disable_index_result_cache = true` under `[scenario.remote_write]` to launch both base and candidate datanodes with Mito's range result cache and index result cache set to `0B`; this avoids range/index result cache hits that bypass the applier; inspect actual EXPLAIN applier metrics to verify the query path. The default is false, so other cases are unchanged.
+
 Large manual remote-write cases can set `sample_chunk_size` to split ingestion by
 time. For each chunk, `prepare-remote` invokes `query_perf_fixture prom-remote-write` with the
 same series cardinality but a shorter `--samples-per-series` and an advanced

tests/perf/query_cases/opt03_opt05_inverted_applier/case.toml

@@ -0,0 +1,88 @@
+# Shared fixture-only A/B case for the OPT03/OPT05 inverted-index applier work.
+# The generic remote-write fixture generator must not hard-code applier behavior;
+# this file only supplies one schema/layout/query instance. Both targets create
+# the same physical metric table with an inverted index and flush every one of
+# the four sample chunks, so query samples exercise the applier over multiple
+# flushed indexed SSTs with the range/index result caches disabled.
+#
+# Manual result inspection (counts are not gated automatically):
+#   host0001 + instance000001 -> 1024
+#   host0001 + instance001025 -> 1024
+#   all rows                  -> 2097152
+#   instance999999            -> 0
+# The EXPLAIN ANALYZE VERBOSE queries are diagnostics for the actual applier
+# metrics; do not read a speedup out of their single timed sample.
+
+[case]
+name = "opt03_opt05_inverted_applier"
+description = "A/B inverted-index applier over flushed indexed Prometheus metric SSTs"
+
+[scenario]
+kind = "prom_remote_write_then_query"
+
+[scenario.remote_write]
+database = "public"
+metric = "opt03_opt05_inverted_applier"
+physical_table = "opt03_opt05_inverted_applier_physical"
+disable_index_result_cache = true
+series_count = 2048
+samples_per_series = 1024
+sample_chunk_size = 256
+flush_every_sample_chunks = 1
+start_unix_millis = 1704067200000
+step_millis = 60000
+chunk_series_count = 256
+timeout_seconds = 600
+visibility_timeout_seconds = 120
+base_setup_sql = [
+  "CREATE TABLE opt03_opt05_inverted_applier_physical (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ('physical_metric_table'='', 'index.type'='inverted', 'compaction.twcs.trigger_file_num'='100')",
+]
+candidate_setup_sql = [
+  "CREATE TABLE opt03_opt05_inverted_applier_physical (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ('physical_metric_table'='', 'index.type'='inverted', 'compaction.twcs.trigger_file_num'='100')",
+]
+
+[scenario.remote_write.prom_store]
+pending_rows_flush_interval = "1s"
+max_batch_rows = 100000
+
+[[scenario.queries]]
+name = "host_and_instance_present"
+kind = "sql"
+query = "SELECT count(*) FROM opt03_opt05_inverted_applier WHERE host = 'host0001' AND instance = 'instance000001'"
+warmup = 3
+iterations = 20
+
+[[scenario.queries]]
+name = "host_and_second_instance_present"
+kind = "sql"
+query = "SELECT count(*) FROM opt03_opt05_inverted_applier WHERE host = 'host0001' AND instance = 'instance001025'"
+warmup = 3
+iterations = 20
+
+[[scenario.queries]]
+name = "explain_host_and_instance_present"
+kind = "sql"
+query = "EXPLAIN ANALYZE VERBOSE SELECT count(*) FROM opt03_opt05_inverted_applier WHERE host = 'host0001' AND instance = 'instance000001'"
+warmup = 0
+iterations = 1
+
+[[scenario.queries]]
+name = "explain_host_and_second_instance_present"
+kind = "sql"
+query = "EXPLAIN ANALYZE VERBOSE SELECT count(*) FROM opt03_opt05_inverted_applier WHERE host = 'host0001' AND instance = 'instance001025'"
+warmup = 0
+iterations = 1
+
+[[scenario.queries]]
+name = "all_rows_control"
+kind = "sql"
+query = "SELECT count(*) FROM opt03_opt05_inverted_applier"
+warmup = 3
+iterations = 9
+
+[[scenario.queries]]
+name = "absent_instance_control"
+kind = "sql"
+query = "SELECT count(*) FROM opt03_opt05_inverted_applier WHERE instance = 'instance999999'"
+warmup = 3
+iterations = 9

tests/perf/test_query_regression_runner_compaction_toctou.py

@@ -39,8 +39,10 @@ class RemoteWriteLifecycleTest(unittest.TestCase):
 
     def test_measure_report_is_finalized_after_datanodes_stop(self) -> None:
         events: list[str] = []
+        commands: list[list[str]] = []
 
         def run(command, **_kwargs):
+            commands.append(command)
             phase = command[1]
             events.append(phase)
             if phase == "measure":
@@ -71,6 +73,13 @@ class RemoteWriteLifecycleTest(unittest.TestCase):
 
         measure = events.index("measure")
         finalize = events.index("finalize-remote")
+        datanode_configs = [command for command in commands if "--target" in command]
+        self.assertEqual(len(datanode_configs), 2)
+        self.assertTrue(all(command[command.index("--target") + 1] == "datanode" for command in datanode_configs))
+        self.assertTrue(all("--config-file" in runner.component_command(target, "datanode") for target in [
+            runner.make_target("base", Path("base"), Path("work"), self.ports[:8], Path("front.toml"), Path("data.toml")),
+            runner.make_target("candidate", Path("candidate"), Path("work"), self.ports[8:], Path("front.toml"), Path("data.toml")),
+        ]))
         self.assertLess(measure, events.index("stop:base:datanode"))
         self.assertLess(events.index("stop:base:datanode"), finalize)
         self.assertLess(events.index("stop:candidate:datanode"), finalize)