From 12f83828b1d4fd6bf9370bb2f5c720ee614b93c6 Mon Sep 17 00:00:00 2001 From: discord9 Date: Tue, 7 Jul 2026 15:58:56 +0800 Subject: [PATCH] feat: add Prom remote-write query regression scenario (#8413) * feat: add Prom remote-write query regression scenario Signed-off-by: discord9 * test: add high-cardinality remote-write query case Signed-off-by: discord9 * feat: chunk remote-write query regression loads Signed-off-by: discord9 * test: use multi-day remote-write regression case Signed-off-by: discord9 * fix: address query regression review comments Signed-off-by: discord9 * ci: allow large query regression comments Signed-off-by: discord9 --------- Signed-off-by: discord9 --- .github/scripts/query-regression-comment.cjs | 9 +- .github/scripts/query-regression-run.py | 9 + .../workflows/query-regression-comment.yml | 6 - .github/workflows/query-regression.yml | 7 +- src/cmd/Cargo.toml | 4 + src/cmd/src/bin/prom_remote_write_fixture.rs | 144 +++++++ src/cmd/src/bin/query_perf_fixture.rs | 48 ++- tests/perf/README.md | 102 ++++- tests/perf/fixture-format.md | 7 + .../prom_remote_write_7913/case.toml | 61 +++ .../prom_remote_write_smoke/case.toml | 31 ++ .../promql_pushdown_7913/case.toml | 54 ++- tests/perf/query_regression_runner.py | 384 ++++++++++++++++-- 13 files changed, 791 insertions(+), 75 deletions(-) create mode 100644 src/cmd/src/bin/prom_remote_write_fixture.rs create mode 100644 tests/perf/query_cases/prom_remote_write_7913/case.toml create mode 100644 tests/perf/query_cases/prom_remote_write_smoke/case.toml diff --git a/.github/scripts/query-regression-comment.cjs b/.github/scripts/query-regression-comment.cjs index 100b1d12d2..5a98e0b816 100644 --- 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'); diff --git a/.github/scripts/query-regression-run.py b/.github/scripts/query-regression-run.py index 4714b5dd0f..b89857be81 100644 --- 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")]) diff --git a/.github/workflows/query-regression-comment.yml b/.github/workflows/query-regression-comment.yml index b320f78e17..21592f6fbc 100644 --- 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)); diff --git a/.github/workflows/query-regression.yml b/.github/workflows/query-regression.yml index 2f11462d09..a898be6ad2 100644 --- 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 diff --git a/src/cmd/Cargo.toml b/src/cmd/Cargo.toml index a11db11962..a8adfa01a4 100644 --- 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", diff --git a/src/cmd/src/bin/prom_remote_write_fixture.rs b/src/cmd/src/bin/prom_remote_write_fixture.rs new file mode 100644 index 0000000000..0bfdf95482 --- /dev/null +++ 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> { + 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::>(); + rows += timeseries + .iter() + .map(|ts| ts.samples.len() as u64) + .sum::(); + 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) +} diff --git a/src/cmd/src/bin/query_perf_fixture.rs b/src/cmd/src/bin/query_perf_fixture.rs index 911bca32ee..dfc3a664a0 100644 --- 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::::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::>(), )) 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::>(), )) as ArrayRef) } @@ -382,11 +394,7 @@ fn generate_record_batch( let mut pk_builder = BinaryDictionaryBuilder::::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"); } diff --git a/tests/perf/README.md b/tests/perf/README.md index 8d82e3a1c3..a59081b568 100644 --- 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('')` 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 ` for slow lab runs. +For large direct-SST fixtures, pass `--fixture-cache-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: diff --git a/tests/perf/fixture-format.md b/tests/perf/fixture-format.md index 0086edf627..ca60ae54d7 100644 --- 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. diff --git a/tests/perf/query_cases/prom_remote_write_7913/case.toml b/tests/perf/query_cases/prom_remote_write_7913/case.toml new file mode 100644 index 0000000000..99a8585ddb --- /dev/null +++ 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 diff --git a/tests/perf/query_cases/prom_remote_write_smoke/case.toml b/tests/perf/query_cases/prom_remote_write_smoke/case.toml new file mode 100644 index 0000000000..b946fbc8f5 --- /dev/null +++ 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 diff --git a/tests/perf/query_cases/promql_pushdown_7913/case.toml b/tests/perf/query_cases/promql_pushdown_7913/case.toml index de6f73f53c..decda960d6 100644 --- 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 diff --git a/tests/perf/query_regression_runner.py b/tests/perf/query_regression_runner.py index 176e51f2b8..a7f2646ea8 100644 --- 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 = []