diff --git a/Cargo.lock b/Cargo.lock index e699780f8f..38927a191a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -13635,6 +13635,7 @@ dependencies = [ "local-ip-address", "mysql", "num_cpus", + "regex", "reqwest 0.12.24", "serde", "serde_json", @@ -13646,6 +13647,7 @@ dependencies = [ "tokio", "tokio-postgres", "tokio-stream", + "toml 0.8.23", ] [[package]] diff --git a/tests/compatibility/AGENTS.md b/tests/compatibility/AGENTS.md new file mode 100644 index 0000000000..edce1edf1a --- /dev/null +++ b/tests/compatibility/AGENTS.md @@ -0,0 +1,53 @@ +# Agent Guidance for Compatibility Framework + +This file is intended for AI agents editing compat cases or the compat runner. Follow these rules to avoid common pitfalls. + +## `verify.result` is Expected Output (Not Auto-Generated Silently) + +- `verify.result` contains the **expected** output of `verify.sql`. +- If the file is **missing**, the runner generates it from actual output and **fails**. + The agent must review the generated file (`git diff`), hand-verify correctness, + and commit it before rerunning. Do **not** blindly commit generated output. +- If actual output **differs** from the expected file, the runner **updates** + `verify.result` with actual output and **fails**. The agent must inspect the + diff and decide whether the change is intentional (accept) or a bug (fix code). + +## Namespace Isolation + +- Each case owns a **unique** namespace. No shared namespace support exists. +- Duplicate namespaces are a **hard error** detected before version filtering. +- The removed `isolation` field is no longer recognized; `case.toml` uses + `deny_unknown_fields`, so any stale `isolation = "shared"` entry causes a + parse error. + +## `case.toml` is Strict + +- `deny_unknown_fields` is enabled. Unknown keys cause a hard parse error. +- Version constraint entries in `from_range` / `to_range` are validated early; + invalid constraints (e.g. `>=not-a-version`) are hard errors, not silent skips. +- All required fields must be non-empty: `name`, `reason`, `introduced_by`, + `topologies`, `from_range`, `to_range`, `features`, `owner`. + +## Phase Semantics + +- `setup.sql` runs on the **old (from)** binary. Only success is required; + output is not compared to any file. +- `verify.sql` runs on the **new (to)** binary. Output is compared against + `verify.result`. + +## PostgreSQL Protocol Cases + +- When using `-- SQLNESS PROTOCOL POSTGRES`, avoid unqualified table names + starting with `pg_`. GreptimeDB issue #8359 causes the parser to rewrite them + to `pg_catalog.`. Qualify such names explicitly or rename the table. + +## Previewing with `--dry-run` + +```shell +cargo run -p sqlness-runner -- compat --dry-run [--from-version vX.Y.Z] [--test-filter "..."] +``` + +The dry-run performs full discovery and filtering (name, topology, metadata +validation, namespace dedup, version-range matching) but starts no services, +creates no temp dirs, and mutates no files. Use it to check which cases +would be selected before a real run. diff --git a/tests/compatibility/README.md b/tests/compatibility/README.md new file mode 100644 index 0000000000..1b16fdb470 --- /dev/null +++ b/tests/compatibility/README.md @@ -0,0 +1,148 @@ +# GreptimeDB Compatibility Test Framework + +Compatibility tests verify that a newer version of GreptimeDB can read data written by an older version. + +Tests are run via `cargo sqlness compat` and reuse the sqlness-runner infrastructure. + +## Quick Start + +```shell +# Self-compat smoke test (current binary only): +cargo run -p sqlness-runner -- compat + +# Test from a specific released version to current: +cargo run -p sqlness-runner -- compat --from-version v0.9.5 + +# Test between two local binary directories: +cargo run -p sqlness-runner -- compat --from-bins-dir ./bins/old --to-bins-dir ./bins/new + +# Run a specific case: +cargo run -p sqlness-runner -- compat --test-filter "basic_table" + +# Preview which cases would run (no services started): +cargo run -p sqlness-runner -- compat --dry-run --from-version v0.9.5 + +# See all options: +cargo run -p sqlness-runner -- compat --help +``` + +## Prerequisites + +- **Docker** (for etcd): PR1 always uses Docker etcd for distributed metadata. External metadata stores are future work. +- **From binary**: Either `--from-version ` to auto-pull a release, or `--from-bins-dir ` to use a local build. The binary `greptime` must exist directly inside the given directory. +- **To binary**: Defaults to the current debug build (`target/debug/greptime`). Override with `--to-bins-dir `. +- **Custom target-dir**: If you use a non-default `CARGO_TARGET_DIR`, the debug binary won't be at `target/debug/greptime`. Pass `--from-bins-dir` / `--to-bins-dir` explicitly pointing to your custom target directory. Alternatively, run `cargo build -p greptime` without a custom target-dir. + +## Case Format + +Each compat case is a directory under `tests/compatibility/cases/` containing three required files plus an expected output file: + +``` +my_case/ + case.toml # Metadata (required) + setup.sql # SQL to run on old version (required) + verify.sql # SQL to run on new version (required) + verify.result # Expected output from verify.sql +``` + +### `case.toml` — Required Metadata + +```toml +name = "my_case" +reason = "Why this compatibility case exists" +introduced_by = "PR #1234 or feature name" +topologies = ["distributed"] # full distributed topology, including flownode +from_range = ["*"] +to_range = ["*"] +features = ["table"] +owner = "team-name" +# optional: +namespace = "my_explicit_namespace" # defaults to sanitized directory name +``` + +**Required fields**: `name`, `reason`, `introduced_by`, `topologies`, `from_range`, `to_range`, `features`, `owner`. + +### Version-Range Filtering + +`from_range` and `to_range` control which binary versions a case applies to: + +| Entry | Meaning | +|-------|---------| +| `"*"` | Matches any version (including unknown). | +| `"vX.Y.Z"` or `"=vX.Y.Z"` | Matches exactly version X.Y.Z. | +| `">=vX.Y.Z"` | Matches X.Y.Z or later. | +| `">vX.Y.Z"` | Matches versions strictly later than X.Y.Z. | +| `"<=vX.Y.Z"` | Matches X.Y.Z or earlier. | +| `" --version` to infer the version. +- When the version **cannot** be determined (e.g. binary missing or `--version` fails), non-wildcard ranges are **skipped** with a message; wildcard (`*`) ranges still match. + +**Example** (`legacy_jsonb`): +```toml +from_range = ["<=v1.1.0"] +to_range = [">=v1.1.1"] +``` +This case only runs when the old binary is <= v1.1.0 and the new binary is >= v1.1.1. + +### `setup.sql` — Setup Phase (Old Version) + +SQL statements executed on the **old** version cluster. These must succeed (any error fails the case). Setup output is NOT compared against any result file. + +Rules: +- Statements are semicolon-terminated +- `--` prefix for ordinary comments +- `-- SQLNESS ...` interceptor comments follow ordinary sqlness semantics +### `verify.sql` — Verify Phase (New Version) + +SQL statements executed on the **new** version cluster. Output is compared against `verify.result` in sqlness snapshot style. + +### `verify.result` — Expected Output + +Expected output in sqlness format. If this file is missing, the runner generates it from actual output and **fails** — the author must review, commit the generated file, and rerun. + +``` +; + + + +; + + + +``` + +If output differs from expected, the run fails and `verify.result` is updated with actual output. + +## PR1 Limitations + +- **Sqlness interceptors**: `-- SQLNESS ...` comments are applied per statement using the same interceptor registry as the ordinary sqlness runner, including the GreptimeDB `PROTOCOL` interceptor. For `PROTOCOL POSTGRES`, the namespace prelude uses `SET search_path` instead of `USE`. Avoid unqualified PostgreSQL-protocol table names starting with `pg_`: GreptimeDB's current PostgreSQL compatibility parser rewrites them to `pg_catalog.
`. +- **Full distributed topology**: The compat runner starts 1 metasrv + 3 datanodes + 1 frontend + 1 flownode. +- **No comment-based compat config**: The compat runner does not define extra compatibility configuration in SQL comments; sqlness comments keep their normal sqlness meaning. + +## Namespace Isolation + +Each case runs in its own database namespace to prevent cross-case interference: + +- Default namespace is derived from the case directory name (sanitized to `[a-z][a-z0-9_]*`) +- Override with `namespace` in `case.toml` +- Duplicate namespaces are **rejected** at discovery time (before version filtering) +- Before each statement, the runner executes a namespace prelude (not written to verify.result): `CREATE DATABASE IF NOT EXISTS ` via gRPC; then `USE ` for gRPC/MySQL statements or `SET search_path TO ''` for PostgreSQL statements. + +## Batch Behavior + +- All cases in a run share one cluster lifecycle: start old cluster → run all setups → restart with new binary → run all verifies +- Cases run **serially** (no parallelism in PR1). Namespace state is session/protocol state and cannot be shared concurrently. +- Same namespace across cases is rejected. + +## xfail Policy (Future) + +For PR1, all cases are expected to pass. Future PRs will add `xfail` support with required `issue` and `expiry` fields. + +## Cross-Job Distributed State + +PR1 runs setup and verify in the **same job** (same process). Cross-job artifact restore for distributed state is not supported in PR1 due to port randomization and etcd lease expiration. diff --git a/tests/compatibility/cases/basic_table/case.toml b/tests/compatibility/cases/basic_table/case.toml new file mode 100644 index 0000000000..dbe4c76c4f --- /dev/null +++ b/tests/compatibility/cases/basic_table/case.toml @@ -0,0 +1,8 @@ +name = "basic_table" +reason = "Verify basic table create/insert/alter/select compatibility across versions." +introduced_by = "PR1 MVP" +topologies = ["distributed"] +from_range = ["*"] +to_range = ["*"] +features = ["table"] +owner = "team" diff --git a/tests/compatibility/cases/basic_table/setup.sql b/tests/compatibility/cases/basic_table/setup.sql new file mode 100644 index 0000000000..fcd38dc77a --- /dev/null +++ b/tests/compatibility/cases/basic_table/setup.sql @@ -0,0 +1,13 @@ +CREATE TABLE foo(ts TIMESTAMP TIME INDEX, s STRING PRIMARY KEY, i INT); + +INSERT INTO foo VALUES +("2024-02-02 01:00:00+0800", "my_tag_1", 1), +("2024-02-02 02:00:00+0800", "my_tag_2", 2), +("2024-02-02 03:00:00+0800", "my_tag_3", 3); + +ALTER TABLE foo ADD COLUMN f FLOAT; + +INSERT INTO foo VALUES +("2024-02-02 04:00:00+0800", "my_tag_4", 4, 4.4), +("2024-02-02 05:00:00+0800", "my_tag_5", 5, 5.5), +("2024-02-02 06:00:00+0800", "my_tag_6", 6, 6.6); diff --git a/tests/compatibility/cases/basic_table/verify.result b/tests/compatibility/cases/basic_table/verify.result new file mode 100644 index 0000000000..1b8188849c --- /dev/null +++ b/tests/compatibility/cases/basic_table/verify.result @@ -0,0 +1,12 @@ +SELECT ts, i, s, f FROM foo ORDER BY ts; + ++---------------------+---+----------+-----+ +| ts | i | s | f | ++---------------------+---+----------+-----+ +| 2024-02-01T17:00:00 | 1 | my_tag_1 | | +| 2024-02-01T18:00:00 | 2 | my_tag_2 | | +| 2024-02-01T19:00:00 | 3 | my_tag_3 | | +| 2024-02-01T20:00:00 | 4 | my_tag_4 | 4.4 | +| 2024-02-01T21:00:00 | 5 | my_tag_5 | 5.5 | +| 2024-02-01T22:00:00 | 6 | my_tag_6 | 6.6 | ++---------------------+---+----------+-----+ diff --git a/tests/compatibility/cases/basic_table/verify.sql b/tests/compatibility/cases/basic_table/verify.sql new file mode 100644 index 0000000000..c109c5f73e --- /dev/null +++ b/tests/compatibility/cases/basic_table/verify.sql @@ -0,0 +1 @@ +SELECT ts, i, s, f FROM foo ORDER BY ts; diff --git a/tests/compatibility/cases/legacy_jsonb/case.toml b/tests/compatibility/cases/legacy_jsonb/case.toml new file mode 100644 index 0000000000..b6f2982291 --- /dev/null +++ b/tests/compatibility/cases/legacy_jsonb/case.toml @@ -0,0 +1,8 @@ +name = "legacy_jsonb" +reason = "Verify legacy JSONB data written by old binaries can be read by newer binaries without entering JSON2 structured alignment paths." +introduced_by = "PR #8323" +topologies = ["distributed"] +from_range = ["<=v1.1.0"] +to_range = [">=v1.1.1"] +features = ["json", "table"] +owner = "query" diff --git a/tests/compatibility/cases/legacy_jsonb/setup.sql b/tests/compatibility/cases/legacy_jsonb/setup.sql new file mode 100644 index 0000000000..79301725d9 --- /dev/null +++ b/tests/compatibility/cases/legacy_jsonb/setup.sql @@ -0,0 +1,12 @@ +CREATE TABLE legacy_jsonb ( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY, + j JSON +); + +INSERT INTO legacy_jsonb (ts, host, j) VALUES + ('2024-01-01 00:00:01+0000', 'r1', parse_json('{"a":10,"msg":"old-jsonb"}')), + ('2024-01-01 00:00:02+0000', 'r2', parse_json('{"a":20,"msg":"old-jsonb"}')), + ('2024-01-01 00:00:03+0000', 'r3', parse_json('{"msg":"no-a"}')); + +ADMIN FLUSH_TABLE('legacy_jsonb'); diff --git a/tests/compatibility/cases/legacy_jsonb/verify.result b/tests/compatibility/cases/legacy_jsonb/verify.result new file mode 100644 index 0000000000..19472aea03 --- /dev/null +++ b/tests/compatibility/cases/legacy_jsonb/verify.result @@ -0,0 +1,39 @@ +SELECT ts, host, json_to_string(j) AS j +FROM legacy_jsonb +ORDER BY ts, host; + ++---------------------+------+----------------------------+ +| ts | host | j | ++---------------------+------+----------------------------+ +| 2024-01-01T00:00:01 | r1 | {"a":10,"msg":"old-jsonb"} | +| 2024-01-01T00:00:02 | r2 | {"a":20,"msg":"old-jsonb"} | +| 2024-01-01T00:00:03 | r3 | {"msg":"no-a"} | ++---------------------+------+----------------------------+ + +SELECT host, json_to_string(j) AS j +FROM legacy_jsonb +WHERE json_get_int(j, 'a') = 10 +ORDER BY host; + ++------+----------------------------+ +| host | j | ++------+----------------------------+ +| r1 | {"a":10,"msg":"old-jsonb"} | ++------+----------------------------+ + +SHOW CREATE TABLE legacy_jsonb; + ++--------------+---------------------------------------------+ +| Table | Create Table | ++--------------+---------------------------------------------+ +| legacy_jsonb | CREATE TABLE IF NOT EXISTS "legacy_jsonb" ( | +| | "ts" TIMESTAMP(3) NOT NULL, | +| | "host" STRING NULL, | +| | "j" JSON NULL, | +| | TIME INDEX ("ts"), | +| | PRIMARY KEY ("host") | +| | ) | +| | | +| | ENGINE=mito | +| | | ++--------------+---------------------------------------------+ diff --git a/tests/compatibility/cases/legacy_jsonb/verify.sql b/tests/compatibility/cases/legacy_jsonb/verify.sql new file mode 100644 index 0000000000..83de54d384 --- /dev/null +++ b/tests/compatibility/cases/legacy_jsonb/verify.sql @@ -0,0 +1,10 @@ +SELECT ts, host, json_to_string(j) AS j +FROM legacy_jsonb +ORDER BY ts, host; + +SELECT host, json_to_string(j) AS j +FROM legacy_jsonb +WHERE json_get_int(j, 'a') = 10 +ORDER BY host; + +SHOW CREATE TABLE legacy_jsonb; diff --git a/tests/compatibility/cases/merge_mode/case.toml b/tests/compatibility/cases/merge_mode/case.toml new file mode 100644 index 0000000000..357bf2de9d --- /dev/null +++ b/tests/compatibility/cases/merge_mode/case.toml @@ -0,0 +1,8 @@ +name = "merge_mode" +reason = "Verify data written with merge_mode='last_non_null' by old binaries is still merged correctly by newer binaries." +introduced_by = "PR #8334" +topologies = ["distributed"] +from_range = ["*"] +to_range = ["*"] +features = ["table", "merge"] +owner = "query" diff --git a/tests/compatibility/cases/merge_mode/setup.sql b/tests/compatibility/cases/merge_mode/setup.sql new file mode 100644 index 0000000000..73bf924f18 --- /dev/null +++ b/tests/compatibility/cases/merge_mode/setup.sql @@ -0,0 +1,19 @@ +CREATE TABLE merge_mode( + host STRING, + ts TIMESTAMP TIME INDEX, + cpu DOUBLE, + memory DOUBLE, + PRIMARY KEY(host) +) +ENGINE=mito +WITH('merge_mode'='last_non_null'); + +INSERT INTO merge_mode VALUES ('host1', 0, 0, NULL), ('host2', 1, NULL, 1); + +INSERT INTO merge_mode VALUES ('host1', 0, NULL, 10), ('host2', 1, 11, NULL); + +INSERT INTO merge_mode VALUES ('host1', 0, 20, NULL); + +INSERT INTO merge_mode VALUES ('host1', 0, NULL, NULL); + +ADMIN FLUSH_TABLE('merge_mode'); diff --git a/tests/compatibility/cases/merge_mode/verify.result b/tests/compatibility/cases/merge_mode/verify.result new file mode 100644 index 0000000000..2978d858aa --- /dev/null +++ b/tests/compatibility/cases/merge_mode/verify.result @@ -0,0 +1,41 @@ +SELECT host, ts, cpu, memory +FROM merge_mode +ORDER BY host, ts; + ++-------+-------------------------+------+--------+ +| host | ts | cpu | memory | ++-------+-------------------------+------+--------+ +| host1 | 1970-01-01T00:00:00 | 20.0 | 10.0 | +| host2 | 1970-01-01T00:00:00.001 | 11.0 | 1.0 | ++-------+-------------------------+------+--------+ + +SELECT host, cpu +FROM merge_mode +WHERE memory >= 10 +ORDER BY host; + ++-------+------+ +| host | cpu | ++-------+------+ +| host1 | 20.0 | ++-------+------+ + +SHOW CREATE TABLE merge_mode; + ++------------+-------------------------------------------+ +| Table | Create Table | ++------------+-------------------------------------------+ +| merge_mode | CREATE TABLE IF NOT EXISTS "merge_mode" ( | +| | "host" STRING NULL, | +| | "ts" TIMESTAMP(3) NOT NULL, | +| | "cpu" DOUBLE NULL, | +| | "memory" DOUBLE NULL, | +| | TIME INDEX ("ts"), | +| | PRIMARY KEY ("host") | +| | ) | +| | | +| | ENGINE=mito | +| | WITH( | +| | merge_mode = 'last_non_null' | +| | ) | ++------------+-------------------------------------------+ diff --git a/tests/compatibility/cases/merge_mode/verify.sql b/tests/compatibility/cases/merge_mode/verify.sql new file mode 100644 index 0000000000..3f214b18d0 --- /dev/null +++ b/tests/compatibility/cases/merge_mode/verify.sql @@ -0,0 +1,10 @@ +SELECT host, ts, cpu, memory +FROM merge_mode +ORDER BY host, ts; + +SELECT host, cpu +FROM merge_mode +WHERE memory >= 10 +ORDER BY host; + +SHOW CREATE TABLE merge_mode; diff --git a/tests/conf/metasrv-test.toml.template b/tests/conf/metasrv-test.toml.template index 64ceb686ac..cf2c0a20a6 100644 --- a/tests/conf/metasrv-test.toml.template +++ b/tests/conf/metasrv-test.toml.template @@ -3,9 +3,6 @@ heartbeat_interval = "1s" {{ if use_etcd }} ## Store server address default to etcd store. store_addrs = [{store_addrs | unescaped}] - -## The datastore for meta server. -backend = "EtcdStore" {{ endif }} [wal] {{ if is_raft_engine }} diff --git a/tests/runner/Cargo.toml b/tests/runner/Cargo.toml index 75b62f6ff7..9801f6bc0b 100644 --- a/tests/runner/Cargo.toml +++ b/tests/runner/Cargo.toml @@ -21,6 +21,7 @@ hex = "0.4" local-ip-address = "0.6" mysql = { version = "26", default-features = false, features = ["minimal", "rustls-tls"] } num_cpus = "1.16" +regex.workspace = true reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "stream"] } serde.workspace = true serde_json.workspace = true @@ -32,3 +33,4 @@ tinytemplate = "1.2" tokio.workspace = true tokio-postgres = { workspace = true } tokio-stream.workspace = true +toml.workspace = true diff --git a/tests/runner/src/cmd.rs b/tests/runner/src/cmd.rs index f7aaacfc73..e17d910b85 100644 --- a/tests/runner/src/cmd.rs +++ b/tests/runner/src/cmd.rs @@ -13,12 +13,15 @@ // limitations under the License. pub(crate) mod bare; +pub(crate) mod compat; +pub(crate) mod compat_case; pub(crate) mod kube; use std::path::PathBuf; use bare::BareCommand; use clap::Parser; +use compat::CompatCommand; use kube::KubeCommand; #[derive(Parser)] @@ -31,6 +34,7 @@ pub struct Command { #[derive(Parser)] pub enum SubCommand { Bare(BareCommand), + Compat(CompatCommand), Kube(KubeCommand), } diff --git a/tests/runner/src/cmd/compat.rs b/tests/runner/src/cmd/compat.rs new file mode 100644 index 0000000000..b73f7b23e3 --- /dev/null +++ b/tests/runner/src/cmd/compat.rs @@ -0,0 +1,878 @@ +// 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. + +use std::path::PathBuf; +use std::sync::Arc; + +use clap::Parser; +use sqlness::QueryContext; +use sqlness::interceptor::template::DELIMITER as TEMPLATE_DELIMITER; +use sqlness::interceptor::{InterceptorRef, Registry}; + +use crate::cmd::bare::ServerAddr; +use crate::cmd::compat_case::{self, CompatCase, try_infer_version, version_matches_range}; +use crate::env::bare::{Env, StoreConfig, WalConfig}; +use crate::protocol_interceptor::{self, POSTGRES, PROTOCOL_KEY}; +use crate::util; + +const COMPAT_TOPOLOGY: &str = "distributed"; +const COMMENT_PREFIX: &str = "--"; +const INTERCEPTOR_PREFIX: &str = "-- SQLNESS"; +const QUERY_DELIMITER: char = ';'; + +/// Run compatibility tests in bare distributed mode. +/// +/// Starts an old-version distributed cluster, runs setup SQLs, +/// then restarts the cluster with a new version on preserved state, +/// and runs verify SQLs comparing results against `verify.result` files. +/// +/// PR1 notes: +/// - Sqlness interceptor comments are supported for each statement. +/// - The runner starts the full distributed topology, including flownode. +#[derive(Debug, Parser)] +pub struct CompatCommand { + /// Version of the "from" GreptimeDB binary (e.g. "v0.9.5") or "current". + /// If neither --from-version nor --from-bins-dir is specified, the + /// current debug build is used for both from and to. + #[clap(long)] + from_version: Option, + + /// Path to the directory containing the "from" GreptimeDB binary. + #[clap(long)] + from_bins_dir: Option, + + /// Path to the directory containing the "to" GreptimeDB binary. + /// Defaults to the current debug build. + #[clap(long)] + to_bins_dir: Option, + + /// Directory of compatibility test cases. + /// Defaults to `tests/compatibility/cases` relative to workspace root. + #[clap(long)] + case_dir: Option, + + /// Name of test cases to run. Accepts a regexp. + #[clap(long, default_value = ".*")] + test_filter: String, + + /// Fail this run as soon as one case fails. + #[clap(long, default_value = "false")] + fail_fast: bool, + + /// Preserve persistent state in the temporary directory after run. + /// Etcd is always cleaned up regardless of this flag. + #[clap(long, default_value = "false")] + preserve_state: bool, + + /// Pull different versions of GreptimeDB on need. + #[clap(long, default_value = "true")] + pull_version_on_need: bool, + + /// Whether to set up etcd via Docker. Required for PR1 distributed compat. + /// External metadata stores are not supported by the compat MVP yet. + #[clap(long, default_value = "true")] + setup_etcd: bool, + + /// Perform discovery and filtering only; print what would run without + /// starting any services, mutating files, or running setup/verify. + #[clap(long, default_value = "false")] + dry_run: bool, +} + +impl CompatCommand { + pub async fn run(self) { + let dry_run = self.dry_run; + + // ---- 1. Validate MVP runtime constraints ---- + if !dry_run && !self.setup_etcd { + panic!( + "compat MVP requires Docker etcd (--setup-etcd=true); external metadata stores are not supported yet" + ); + } + + // ---- 2. Resolve case directory ---- + let case_dir = self.case_dir.unwrap_or_else(default_compat_case_dir); + + if !case_dir.is_dir() { + panic!("Case directory not found: {}", case_dir.display()); + } + + // ---- 3. Discover cases ---- + let mut cases = compat_case::discover_cases(&case_dir).unwrap_or_else(|e| panic!("{e}")); + + // Filter by test_filter + let filter_re = regex::Regex::new(&self.test_filter) + .unwrap_or_else(|e| panic!("Invalid test filter regex '{}': {e}", self.test_filter)); + cases.retain(|c| filter_re.is_match(&c.metadata.name)); + + // Filter by topology + cases.retain(|c| c.metadata.topologies.iter().any(|t| t == COMPAT_TOPOLOGY)); + + if cases.is_empty() { + if dry_run { + println!( + "DRY-RUN: no compat cases found matching filter '{}' and topology '{}'", + self.test_filter, COMPAT_TOPOLOGY + ); + } else { + println!( + "No compat cases found matching filter '{}' and topology '{}'", + self.test_filter, COMPAT_TOPOLOGY + ); + } + return; + } + + // ---- 3b. Validate metadata (incl. version constraints) before filtering ---- + // Must run before version-range filtering so invalid constraints like + // `>=not-a-version` cause a hard error instead of silent skip. + compat_case::validate_cases_metadata(&cases).unwrap_or_else(|e| panic!("{e}")); + + // ---- 3c. Validate namespace dedup before version filtering ---- + // Validate globally for all selected topology/name cases so duplicated + // namespaces cannot hide behind version filters. + compat_case::validate_case_namespaces(&cases).unwrap_or_else(|e| panic!("{e}")); + + // ---- 4. Resolve "from" and "to" versions ---- + let (from_bins_dir, from_version, from_ver_parsed, to_bins_dir, to_version, to_ver_parsed) = + if dry_run { + // Dry-run: resolve versions without panicking on missing binaries. + // try_infer_version returns None gracefully when the binary is absent. + let dry_run_from_bins_dir = self + .from_bins_dir + .clone() + .unwrap_or_else(|| util::get_binary_dir("debug")); + let from_ver_str = self + .from_version + .as_deref() + .and_then(|v| { + if v == "current" { + None + } else { + Some(v.to_string()) + } + }) + .or_else(|| try_infer_version(&dry_run_from_bins_dir).map(|v| v.to_string())); + let from_ver_parsed = from_ver_str + .as_deref() + .and_then(|s| compat_case::Version::parse(s).ok()); + + let dry_run_to_bins_dir = self + .to_bins_dir + .clone() + .unwrap_or_else(|| util::get_binary_dir("debug")); + let to_ver_str = try_infer_version(&dry_run_to_bins_dir).map(|v| v.to_string()); + let to_ver_parsed = to_ver_str + .as_deref() + .and_then(|s| compat_case::Version::parse(s).ok()); + + ( + Some(dry_run_from_bins_dir), + from_ver_str, + from_ver_parsed, + Some(dry_run_to_bins_dir), + to_ver_str, + to_ver_parsed, + ) + } else { + // Normal path: resolve bins (may panic if binary not found). + let from_bins_dir = resolve_bins( + self.from_bins_dir.as_ref(), + self.from_version.as_deref(), + self.pull_version_on_need, + ) + .await; + + let from_version = if let Some(ref ver) = self.from_version { + if ver != "current" { + Some(ver.clone()) + } else { + try_infer_version(&from_bins_dir).map(|v| v.to_string()) + } + } else { + try_infer_version(&from_bins_dir).map(|v| v.to_string()) + }; + + let from_ver_parsed = from_version + .as_deref() + .and_then(|s| compat_case::Version::parse(s).ok()); + + let to_bins_dir = + resolve_bins(self.to_bins_dir.as_ref(), None, self.pull_version_on_need).await; + let to_version = try_infer_version(&to_bins_dir).map(|v| v.to_string()); + let to_ver_parsed = to_version + .as_deref() + .and_then(|s| compat_case::Version::parse(s).ok()); + + ( + Some(from_bins_dir), + from_version, + from_ver_parsed, + Some(to_bins_dir), + to_version, + to_ver_parsed, + ) + }; + + // ---- 5b. Filter by version range ---- + let pre_filter_count = cases.len(); + cases.retain(|c| { + let from_ok = version_matches_range(from_ver_parsed.as_ref(), &c.metadata.from_range); + if !from_ok { + let from_label = from_ver_parsed + .as_ref() + .map(|v| v.to_string()) + .unwrap_or_else(|| "unknown".to_string()); + println!( + "Skipping case '{}': from_range {:?} does not match version '{}'", + c.metadata.name, c.metadata.from_range, from_label + ); + } + from_ok + }); + cases.retain(|c| { + let to_ok = version_matches_range(to_ver_parsed.as_ref(), &c.metadata.to_range); + if !to_ok { + let to_label = to_ver_parsed + .as_ref() + .map(|v| v.to_string()) + .unwrap_or_else(|| "unknown".to_string()); + println!( + "Skipping case '{}': to_range {:?} does not match version '{}'", + c.metadata.name, c.metadata.to_range, to_label + ); + } + to_ok + }); + + if pre_filter_count != cases.len() { + println!( + "Version-range filtering: {} → {} cases", + pre_filter_count, + cases.len() + ); + } + + if cases.is_empty() { + if dry_run { + println!("DRY-RUN: no compat cases would run after version-range filtering"); + } else { + println!("No compat cases remaining after version-range filtering"); + } + return; + } + + if dry_run { + println!("DRY-RUN: would run {} compat case(s)", cases.len()); + println!(" topology: {}", COMPAT_TOPOLOGY); + println!( + " from version: {}", + from_version.as_deref().unwrap_or( + "unknown (use --from-version, --from-bins-dir, or build debug binary)" + ) + ); + println!( + " to version: {}", + to_version + .as_deref() + .unwrap_or("unknown (use --to-bins-dir or build debug binary)") + ); + if pre_filter_count != cases.len() { + println!(); + println!( + "Version-range filtering reduced {} → {} cases (see 'Skipping case' messages above)", + pre_filter_count, + cases.len() + ); + } + println!(); + for c in &cases { + println!(" case: {}", c.metadata.name); + println!(" namespace: {}", c.namespace); + println!(" from_range: {:?}", c.metadata.from_range); + println!(" to_range: {:?}", c.metadata.to_range); + println!(" features: {:?}", c.metadata.features); + } + println!(); + println!("Dry run complete. Remove --dry-run to execute."); + return; + } + + println!( + "Running {} compat case(s) with topology {}:", + cases.len(), + COMPAT_TOPOLOGY + ); + for c in &cases { + println!( + " - {} (namespace: {}, topologies: {:?})", + c.metadata.name, c.namespace, c.metadata.topologies + ); + } + + // ---- 6. Create temp directory (after filtering so early exits don't leave empty dirs) ---- + let temp_dir = tempfile::Builder::new() + .prefix("sqlness-compat") + .tempdir() + .unwrap(); + let sqlness_home = temp_dir.keep(); + unsafe { + std::env::set_var("SQLNESS_HOME", sqlness_home.display().to_string()); + } + + // ---- 7. Build interceptor registry ---- + let interceptor_registry = create_interceptor_registry(); + + // ---- 7b. Create Env for bare distributed mode ---- + let store_config = StoreConfig { + store_addrs: if self.setup_etcd { + vec!["127.0.0.1:2379".to_string()] + } else { + vec![] + }, + setup_etcd: self.setup_etcd, + setup_pg: None, + setup_mysql: None, + enable_flat_format: false, + }; + + let env = Env::new( + sqlness_home.clone(), + ServerAddr::default(), + WalConfig::RaftEngine, + self.pull_version_on_need, + from_bins_dir, + store_config, + vec![], + ); + + // ---- 7c. Etcd cleanup guard ---- + // Arm this only immediately before starting the cluster. Earlier validation + // failures should not stop an unrelated local container named `etcd`. + let mut etcd_guard = if self.setup_etcd { + Some(EtcdGuard::new()) + } else { + None + }; + + // ---- 8. Run setup phase on old cluster ---- + println!("Starting old-version distributed cluster with flownode..."); + let mut db = env.compat_start_distributed(0).await; + + println!("Running setup phase..."); + for case in &cases { + run_compat_phase(&db, case, &interceptor_registry, CompatPhase::Setup) + .await + .unwrap_or_else(|e| panic!("Setup failed for case '{}': {e}", case.metadata.name)); + println!(" Setup: {} - OK", case.metadata.name); + } + + // ---- 9. Switch to "to" binary and restart cluster ---- + // to_bins_dir was already resolved during version-range filtering + println!("Restarting cluster with new-version binary on preserved state..."); + env.compat_restart_all( + &db, + to_bins_dir.expect("to_bins_dir must be resolved in non-dry-run mode"), + ) + .await; + + // ---- 10. Run verify phase on new cluster ---- + println!("Running verify phase..."); + let mut failed = Vec::new(); + for case in &cases { + match run_compat_phase(&db, case, &interceptor_registry, CompatPhase::Verify).await { + Ok(()) => println!(" Verify: {} - PASSED", case.metadata.name), + Err(e) => { + println!(" Verify: {} - FAILED: {e}", case.metadata.name); + failed.push(case.metadata.name.clone()); + if self.fail_fast { + break; + } + } + } + } + + // ---- 11. Stop cluster ---- + db.compat_stop(); + + // ---- 12. Cleanup ---- + // Etcd is always cleaned up; --preserve-state only preserves sqlness_home. + if self.setup_etcd { + println!("Stopping etcd"); + util::stop_rm_etcd(); + } + + if !self.preserve_state { + println!("Removing state in {:?}", sqlness_home); + tokio::fs::remove_dir_all(sqlness_home) + .await + .unwrap_or_else(|e| println!("Warning: failed to clean up temp dir: {e}")); + } + + // Disarm the etcd guard now that we've done normal cleanup. + if let Some(mut guard) = etcd_guard.take() { + guard.disarm(); + } + + if failed.is_empty() { + println!("\n\x1b[32mAll compat tests passed!\x1b[0m"); + } else { + println!("\n\x1b[31mFailed cases: {}\x1b[0m", failed.join(", ")); + // Explicitly drop the guard before exit so it doesn't double-cleanup. + std::process::exit(1); + } + } +} + +/// Guard that stops/removes Docker etcd on drop (panic or early exit). +/// Disarm before normal cleanup to avoid double-cleanup. +/// +/// The guard refuses to arm if a container named `etcd` already exists, so a +/// failed compat run never deletes a developer-owned container with that name. +struct EtcdGuard { + active: bool, +} + +impl EtcdGuard { + fn new() -> Self { + let inspect_status = std::process::Command::new("docker") + .args(["container", "inspect", "etcd"]) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .status(); + if inspect_status.is_ok_and(|status| status.success()) { + panic!( + "A Docker container named `etcd` already exists. \ + Remove it before running compat tests so the cleanup guard \ + cannot delete a container it did not create." + ); + } + Self { active: true } + } + + fn disarm(&mut self) { + self.active = false; + } +} + +impl Drop for EtcdGuard { + fn drop(&mut self) { + if self.active { + println!("EtcdGuard: emergency etcd cleanup (panic or early exit)"); + // Best-effort: don't panic in Drop + let _ = std::process::Command::new("docker") + .args(["container", "stop", "etcd"]) + .status(); + let _ = std::process::Command::new("docker") + .args(["container", "rm", "etcd"]) + .status(); + } + } +} + +/// Phase of compat execution. +#[derive(Clone, Copy, PartialEq, Eq)] +enum CompatPhase { + Setup, + Verify, +} + +/// Create an interceptor registry matching the ordinary sqlness runner. +fn create_interceptor_registry() -> Registry { + let mut interceptor_registry: Registry = Default::default(); + interceptor_registry.register( + protocol_interceptor::PREFIX, + Arc::new(protocol_interceptor::ProtocolInterceptorFactory), + ); + interceptor_registry +} + +/// Resolve binary directory: explicit path takes priority, then version (pulls if needed), +/// otherwise default to current debug build. +/// +/// Validates that `/greptime` exists after resolution and canonicalizes the path. +async fn resolve_bins( + bins_dir: Option<&PathBuf>, + version: Option<&str>, + pull_version_on_need: bool, +) -> PathBuf { + let dir = if let Some(dir) = bins_dir { + dir.clone() + } else if let Some(ver) = version { + if ver == "current" { + util::get_binary_dir("debug") + } else { + util::maybe_pull_binary(ver, pull_version_on_need).await; + let root = std::path::PathBuf::from(util::get_workspace_root()); + std::path::PathBuf::from_iter([root, std::path::PathBuf::from(ver)]) + } + } else { + // Default: current debug build + util::get_binary_dir("debug") + }; + + // Canonicalize when possible (may fail if dir doesn't exist) + let dir = match dir.canonicalize() { + Ok(canon) => canon, + Err(e) => panic!( + "Cannot resolve binary directory '{}': {e}. \ + Use --from-bins-dir / --to-bins-dir to specify the correct path, \ + or --from-version to pull a release.", + dir.display() + ), + }; + + if !dir.join(util::PROGRAM).is_file() { + panic!( + "greptime binary not found in '{}'. \ + Use --from-bins-dir / --to-bins-dir to specify the correct directory, \ + or build greptime first (e.g. `cargo build -p greptime`). \ + Note: if you use a custom target-dir, the binary may be elsewhere; \ + pass the actual directory with --from-bins-dir or --to-bins-dir.", + dir.display() + ); + } + + dir +} + +/// Default case directory: `tests/compatibility/cases` relative to workspace root. +fn default_compat_case_dir() -> PathBuf { + let mut path = PathBuf::from(env!("CARGO_MANIFEST_DIR")); + // CARGO_MANIFEST_DIR is tests/runner + // Pop to tests/ + path.pop(); + path.push("compatibility"); + path.push("cases"); + path +} + +/// Run a single compat phase (setup or verify) for one case. +async fn run_compat_phase( + db: &crate::env::bare::GreptimeDB, + case: &CompatCase, + registry: &Registry, + phase: CompatPhase, +) -> Result<(), String> { + let sql_file = match phase { + CompatPhase::Setup => case.dir.join("setup.sql"), + CompatPhase::Verify => case.dir.join("verify.sql"), + }; + + let sql_content = std::fs::read_to_string(&sql_file) + .map_err(|e| format!("Failed to read {}: {e}", sql_file.display()))?; + + let mut statements = parse_sql_file(&sql_content, registry)?; + + // Execute statements + let mut verify_output = String::new(); + + for statement in &mut statements { + let (display, results) = statement.execute(db, &case.namespace).await?; + + match phase { + CompatPhase::Setup => { + // Setup: just check for success (already returned Ok) + } + CompatPhase::Verify => { + verify_output.push_str(&display); + for result in results { + verify_output.push_str(&result); + verify_output.push('\n'); + verify_output.push('\n'); + } + } + } + } + + if phase == CompatPhase::Verify { + trim_trailing_blank_lines(&mut verify_output); + + let result_path = case.dir.join("verify.result"); + + // If verify.result doesn't exist, generate it from actual output but + // return an error so the author must review, commit, and rerun. + if !result_path.is_file() { + std::fs::write(&result_path, &verify_output) + .map_err(|e| format!("Failed to create {}: {e}", result_path.display()))?; + return Err(format!( + "Created missing verify.result for case '{}'; review the generated file, commit it, and rerun", + case.metadata.name + )); + } + + let expected = std::fs::read_to_string(&result_path) + .map_err(|e| format!("Failed to read {}: {e}", result_path.display()))?; + + if verify_output != expected { + // Update the result file with actual output to aid local update. + std::fs::write(&result_path, &verify_output) + .map_err(|e| format!("Failed to update {}: {e}", result_path.display()))?; + + // Generate a simple diff + let diff = simple_diff(&expected, &verify_output); + return Err(format!( + "Result mismatch for case '{}'.\nDiff:\n{diff}", + case.metadata.name + )); + } + } + + Ok(()) +} + +/// Keep generated snapshots compatible with `git diff --check` by avoiding a +/// trailing blank line at EOF while preserving the final newline. +fn trim_trailing_blank_lines(output: &mut String) { + while output.ends_with("\n\n") { + output.pop(); + } +} + +/// Execute the namespace prelude (CREATE DATABASE IF NOT EXISTS + USE) for a case. +/// This is NOT written into verify.result. +/// +/// The prelude is protocol-aware: +/// - `CREATE DATABASE` is always sent via gRPC so it works regardless of +/// statement-level protocol directives. +/// - For Postgres-protocol statements, `SET search_path` selects the case +/// namespace instead of running `USE` (which is not valid PG SQL). +/// - For MySQL and default/gRPC statements, `USE ` runs through the +/// statement's effective context. +async fn run_namespace_prelude( + db: &crate::env::bare::GreptimeDB, + namespace: &str, + query_ctx: &QueryContext, +) -> Result<(), String> { + // CREATE DATABASE always via gRPC — no protocol override + let create_db = format!("CREATE DATABASE IF NOT EXISTS {namespace}"); + let default_ctx = QueryContext::default(); + db.compat_query(&create_db, &default_ctx).await?; + + // Postgres: select the namespace via search_path instead of USE. + if query_ctx + .context + .get(PROTOCOL_KEY) + .is_some_and(|p| p == POSTGRES) + { + let set_search_path = format!("SET search_path TO '{namespace}'"); + db.compat_query(&set_search_path, query_ctx).await?; + return Ok(()); + } + + // MySQL / default (gRPC): execute USE + let use_db = format!("USE {namespace}"); + db.compat_query(&use_db, query_ctx).await?; + + Ok(()) +} + +/// A parsed SQL statement with sqlness comments and interceptors. +struct ParsedStatement { + comment_lines: Vec, + display_query: Vec, + execute_query: Vec, + interceptors: Vec, +} + +impl ParsedStatement { + fn new() -> Self { + Self { + comment_lines: Vec::new(), + display_query: Vec::new(), + execute_query: Vec::new(), + interceptors: Vec::new(), + } + } + + fn push_comment(&mut self, line: String) { + self.comment_lines.push(line); + } + + fn push_interceptor(&mut self, line: &str, registry: &Registry) -> Result<(), String> { + let Some((_, remaining)) = line.split_once(INTERCEPTOR_PREFIX) else { + return Err(format!( + "Missing sqlness interceptor prefix in line: {line}" + )); + }; + let interceptor = registry.create(remaining).map_err(|e| e.to_string())?; + self.interceptors.push(interceptor); + Ok(()) + } + + fn append_query_line(&mut self, line: &str) { + self.display_query.push(line.to_string()); + self.execute_query.push(line.to_string()); + } + + fn is_empty(&self) -> bool { + self.comment_lines.is_empty() + && self.display_query.is_empty() + && self.execute_query.is_empty() + && self.interceptors.is_empty() + } + + fn has_query(&self) -> bool { + !self.execute_query.is_empty() + } + + fn display_text(&self) -> String { + let mut output = String::new(); + for comment in &self.comment_lines { + output.push_str(comment); + output.push('\n'); + } + for line in &self.display_query { + output.push_str(line); + } + output.push('\n'); + output.push('\n'); + output + } + + fn concat_query_lines(&self) -> String { + self.execute_query + .iter() + .fold(String::new(), |query, line| query + line) + .trim_start() + .to_string() + } + + async fn before_execute_intercept(&mut self) -> QueryContext { + let mut context = QueryContext::default(); + for interceptor in &self.interceptors { + interceptor + .before_execute_async(&mut self.execute_query, &mut context) + .await; + } + context + } + + async fn after_execute_intercept(&self, result: &mut String) { + for interceptor in &self.interceptors { + interceptor.after_execute_async(result).await; + } + } + + async fn execute( + &mut self, + db: &crate::env::bare::GreptimeDB, + namespace: &str, + ) -> Result<(String, Vec), String> { + let display = self.display_text(); + let context = self.before_execute_intercept().await; + db.compat_prepare_query_context(&context).await; + run_namespace_prelude(db, namespace, &context).await?; + let sql = self.concat_query_lines(); + let mut results = Vec::new(); + + for sql in sql.split(TEMPLATE_DELIMITER) { + if sql.trim().is_empty() { + continue; + } + let sql = if sql.ends_with(QUERY_DELIMITER) { + sql.to_string() + } else { + format!("{sql};") + }; + let mut result = db.compat_query(&sql, &context).await?; + self.after_execute_intercept(&mut result).await; + results.push(result); + } + + Ok((display, results)) + } +} + +/// Parse a SQL file into statements using the same sqlness comment/interceptor +/// conventions as the ordinary runner. +fn parse_sql_file(content: &str, registry: &Registry) -> Result, String> { + let mut statements = Vec::new(); + let mut current_stmt = ParsedStatement::new(); + + for line in content.lines() { + if line.starts_with(COMMENT_PREFIX) { + current_stmt.push_comment(line.to_string()); + + if line.starts_with(INTERCEPTOR_PREFIX) { + current_stmt.push_interceptor(line, registry)?; + } + continue; + } + + if line.is_empty() { + continue; + } + + current_stmt.append_query_line(line); + + // Check for statement terminator + if line.ends_with(QUERY_DELIMITER) { + if current_stmt.has_query() { + statements.push(current_stmt); + } + current_stmt = ParsedStatement::new(); + } else { + current_stmt.append_query_line("\n"); + } + } + + // Flush any remaining statement + if !current_stmt.is_empty() && current_stmt.has_query() { + statements.push(current_stmt); + } + + if statements.is_empty() { + return Err("No SQL statements found in file".to_string()); + } + + Ok(statements) +} + +/// Generate a simple line-based diff between expected and actual. +fn simple_diff(expected: &str, actual: &str) -> String { + let mut diff = String::new(); + let expected_lines: Vec<&str> = expected.lines().collect(); + let actual_lines: Vec<&str> = actual.lines().collect(); + let max_len = expected_lines.len().max(actual_lines.len()); + + for i in 0..max_len { + let exp = expected_lines.get(i).unwrap_or(&"(missing)"); + let act = actual_lines.get(i).unwrap_or(&"(missing)"); + if exp != act { + diff.push_str(&format!(" Line {}:\n", i + 1)); + diff.push_str(&format!(" expected: {exp}\n")); + diff.push_str(&format!(" actual: {act}\n")); + } + } + + if diff.is_empty() { + diff.push_str(" (files differ but no line-level diff found — may be whitespace)\n"); + } + + diff +} + +#[cfg(test)] +mod tests { + use super::trim_trailing_blank_lines; + + #[test] + fn test_trim_trailing_blank_lines_preserves_single_final_newline() { + let mut output = "SELECT 1;\n\n+---+\n\n".to_string(); + trim_trailing_blank_lines(&mut output); + + assert_eq!(output, "SELECT 1;\n\n+---+\n"); + } +} diff --git a/tests/runner/src/cmd/compat_case.rs b/tests/runner/src/cmd/compat_case.rs new file mode 100644 index 0000000000..bc9065c449 --- /dev/null +++ b/tests/runner/src/cmd/compat_case.rs @@ -0,0 +1,822 @@ +// 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. + +use std::collections::HashSet; +use std::path::{Path, PathBuf}; +use std::process::Command; + +use serde::Deserialize; + +/// Metadata for a compatibility test case, parsed from `case.toml`. +#[derive(Debug, Clone, Deserialize)] +#[serde(deny_unknown_fields)] +#[allow(dead_code)] +pub struct CaseMetadata { + /// Human-readable name of the case. + pub name: String, + /// Why this compatibility case exists. + pub reason: String, + /// What PR, issue, or feature introduced this case. + pub introduced_by: String, + /// Which topologies this case applies to (e.g. ["distributed"]). + pub topologies: Vec, + /// Version range for the "from" binary. `*` means all versions. + pub from_range: Vec, + /// Version range for the "to" binary. `*` means all versions. + pub to_range: Vec, + /// Features required (e.g. ["table", "flow"]). + pub features: Vec, + /// Owner team or individual. + pub owner: String, + /// Optional explicit namespace. If not set, derived from case directory name. + /// Must match `[a-z0-9_]+`. + #[serde(default)] + pub namespace: Option, +} + +impl CaseMetadata { + /// Compute the effective namespace for this case. + /// Uses explicit `namespace` field if set, otherwise derives from case directory name. + pub fn effective_namespace(&self, case_dir_name: &str) -> String { + self.namespace + .clone() + .unwrap_or_else(|| sanitize_namespace(case_dir_name)) + } +} + +/// A loaded compatibility case (metadata + file paths). +#[derive(Debug, Clone)] +pub struct CompatCase { + /// Parsed metadata from case.toml. + pub metadata: CaseMetadata, + /// Path to the case directory. + pub dir: PathBuf, + /// Effective namespace for this case. + pub namespace: String, +} + +/// Sanitize a name into a valid GreptimeDB namespace: lowercase alphanumeric + underscores. +fn sanitize_namespace(name: &str) -> String { + let sanitized: String = name + .to_lowercase() + .chars() + .map(|c| { + if c.is_ascii_alphanumeric() || c == '_' { + c + } else { + '_' + } + }) + .collect(); + + // Must start with a letter + if sanitized + .chars() + .next() + .is_none_or(|c| !c.is_ascii_alphabetic()) + { + format!("c_{sanitized}") + } else { + sanitized + } +} + +/// Discover all compat cases under `case_root`. +/// Each case is a directory containing `case.toml`, `setup.sql`, and `verify.sql`. +/// `verify.result` is optional at discovery — if missing, the verify phase +/// generates it from actual output and fails so the author must review/commit. +pub fn discover_cases(case_root: &Path) -> Result, String> { + let mut cases = Vec::new(); + + if !case_root.is_dir() { + return Err(format!( + "Case root directory not found: {}", + case_root.display() + )); + } + + let entries = std::fs::read_dir(case_root) + .map_err(|e| format!("Failed to read case root {}: {e}", case_root.display()))?; + + for entry in entries { + let entry = entry.map_err(|e| format!("Failed to read case dir entry: {e}"))?; + let path = entry.path(); + if !path.is_dir() { + continue; + } + + let case_toml_path = path.join("case.toml"); + if !case_toml_path.is_file() { + println!("Skipping directory {}: no case.toml found", path.display()); + continue; + } + + let setup_sql = path.join("setup.sql"); + let verify_sql = path.join("verify.sql"); + + for required in [&setup_sql, &verify_sql] { + if !required.is_file() { + return Err(format!( + "Missing required file {} in case directory {}", + required.display(), + path.display() + )); + } + } + + let content = std::fs::read_to_string(&case_toml_path) + .map_err(|e| format!("Failed to read {}: {e}", case_toml_path.display()))?; + + let metadata: CaseMetadata = toml::from_str(&content) + .map_err(|e| format!("Failed to parse {}: {e}", case_toml_path.display()))?; + + let case_dir_name = path + .file_name() + .and_then(|n| n.to_str()) + .unwrap_or("unknown"); + + let namespace = metadata.effective_namespace(case_dir_name); + + cases.push(CompatCase { + metadata, + dir: path, + namespace, + }); + } + + if cases.is_empty() { + return Err(format!( + "No compat cases found under {}", + case_root.display() + )); + } + + // Sort by directory name for deterministic ordering across runs. + cases.sort_by(|a, b| a.dir.file_name().cmp(&b.dir.file_name())); + + Ok(cases) +} + +/// Validate per-case metadata for all discovered cases. +/// +/// Checks that required fields are non-empty, version constraints are parseable, +/// and namespace format is valid. +/// +/// Call this **before** version-range filtering so that invalid constraints +/// (e.g. `>=not-a-version`) cause a hard error instead of being silently +/// filtered out. +pub fn validate_cases_metadata(cases: &[CompatCase]) -> Result<(), String> { + for case in cases { + // Validate namespace format: must start with a lowercase letter, followed by + // lowercase alphanumeric or underscores only. + if !is_valid_namespace(&case.namespace) { + return Err(format!( + "Case '{}' has invalid namespace '{}': must match [a-z][a-z0-9_]*", + case.metadata.name, case.namespace + )); + } + + // Validate required metadata fields are non-empty + if case.metadata.name.is_empty() { + return Err(format!("Case in {} has empty name", case.dir.display())); + } + if case.metadata.reason.is_empty() { + return Err(format!("Case '{}' has empty reason", case.metadata.name)); + } + if case.metadata.introduced_by.is_empty() { + return Err(format!( + "Case '{}' has empty introduced_by", + case.metadata.name + )); + } + if case.metadata.owner.is_empty() { + return Err(format!("Case '{}' has empty owner", case.metadata.name)); + } + if case.metadata.topologies.is_empty() { + return Err(format!( + "Case '{}' has empty topologies", + case.metadata.name + )); + } + if case.metadata.from_range.is_empty() { + return Err(format!( + "Case '{}' has empty from_range", + case.metadata.name + )); + } + if case.metadata.to_range.is_empty() { + return Err(format!("Case '{}' has empty to_range", case.metadata.name)); + } + validate_version_constraints(&case.metadata.name, "from_range", &case.metadata.from_range)?; + validate_version_constraints(&case.metadata.name, "to_range", &case.metadata.to_range)?; + if case.metadata.features.is_empty() { + return Err(format!("Case '{}' has empty features", case.metadata.name)); + } + } + + Ok(()) +} + +/// Check for duplicate namespaces. +/// +/// Call this **before** version-range filtering so duplicated namespaces cannot +/// hide behind version filters. +pub fn validate_case_namespaces(cases: &[CompatCase]) -> Result<(), String> { + let mut namespaces: HashSet<&str> = HashSet::new(); + + for case in cases { + if !namespaces.insert(&case.namespace) { + return Err(format!( + "Duplicate namespace '{}' for case '{}'. \ + Each case must have a unique effective namespace.", + case.namespace, case.metadata.name + )); + } + } + + Ok(()) +} + +/// Check whether a string is a valid namespace: starts with lowercase letter, +/// contains only lowercase alphanumeric + underscores. +fn is_valid_namespace(s: &str) -> bool { + if s.is_empty() { + return false; + } + let mut chars = s.chars(); + match chars.next() { + Some(c) if c.is_ascii_lowercase() => {} + _ => return false, + } + chars.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_') +} + +fn validate_version_constraints( + case_name: &str, + field_name: &str, + constraints: &[String], +) -> Result<(), String> { + for constraint in constraints { + parse_version_constraint(constraint).map_err(|e| { + format!("Case '{case_name}' has invalid {field_name} entry '{constraint}': {e}") + })?; + } + Ok(()) +} + +// --------------------------------------------------------------------------- +// Version-range filtering +// --------------------------------------------------------------------------- + +/// A simple 3-component version: `major.minor.patch`. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub(crate) struct Version { + pub major: u64, + pub minor: u64, + pub patch: u64, +} + +impl Version { + /// Parse a version string like `v1.1.0` or `1.1.0`. + pub(crate) fn parse(raw: &str) -> Result { + let stripped = raw.strip_prefix('v').unwrap_or(raw); + let core = stripped + .split_once('-') + .map(|(core, _)| core) + .unwrap_or(stripped); + let core = core.split_once('+').map(|(core, _)| core).unwrap_or(core); + let parts: Vec<&str> = core.split('.').collect(); + if parts.len() != 3 { + return Err(format!( + "Invalid version '{}': expected major.minor.patch", + raw + )); + } + let major = parts[0] + .parse::() + .map_err(|_| format!("Invalid major version in '{}'", raw))?; + let minor = parts[1] + .parse::() + .map_err(|_| format!("Invalid minor version in '{}'", raw))?; + let patch = parts[2] + .parse::() + .map_err(|_| format!("Invalid patch version in '{}'", raw))?; + Ok(Self { + major, + minor, + patch, + }) + } +} + +impl std::fmt::Display for Version { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "v{}.{}.{}", self.major, self.minor, self.patch) + } +} + +/// A version constraint used in `from_range` / `to_range`. +#[derive(Debug, Clone)] +enum VersionConstraint { + Wildcard, + Exact(Version), + Gt(Version), + Gte(Version), + Lt(Version), + Lte(Version), +} + +impl VersionConstraint { + /// Does this constraint match the given version? + fn matches(&self, version: &Version) -> bool { + match self { + Self::Wildcard => true, + Self::Exact(target) => version == target, + Self::Gt(target) => version > target, + Self::Gte(target) => version >= target, + Self::Lt(target) => version < target, + Self::Lte(target) => version <= target, + } + } +} + +/// Parse a single range entry (e.g. `*`, `<=v1.1.0`, `>=v1.1.1`, `v1.0.0`). +fn parse_version_constraint(raw: &str) -> Result { + let trimmed = raw.trim(); + if trimmed.is_empty() { + return Err("Empty version constraint".to_string()); + } + if trimmed == "*" { + return Ok(VersionConstraint::Wildcard); + } + + // Ordered from most-specific prefix to least + if let Some(ver_str) = trimmed.strip_prefix(">=") { + let ver = Version::parse(ver_str.trim())?; + return Ok(VersionConstraint::Gte(ver)); + } + if let Some(ver_str) = trimmed.strip_prefix("<=") { + let ver = Version::parse(ver_str.trim())?; + return Ok(VersionConstraint::Lte(ver)); + } + if let Some(ver_str) = trimmed.strip_prefix("==") { + let ver = Version::parse(ver_str.trim())?; + return Ok(VersionConstraint::Exact(ver)); + } + if let Some(ver_str) = trimmed.strip_prefix('=') { + let ver = Version::parse(ver_str.trim())?; + return Ok(VersionConstraint::Exact(ver)); + } + if let Some(ver_str) = trimmed.strip_prefix('>') { + let ver = Version::parse(ver_str.trim())?; + return Ok(VersionConstraint::Gt(ver)); + } + if let Some(ver_str) = trimmed.strip_prefix('<') { + let ver = Version::parse(ver_str.trim())?; + return Ok(VersionConstraint::Lt(ver)); + } + + // No operator → treat as exact + let ver = Version::parse(trimmed)?; + Ok(VersionConstraint::Exact(ver)) +} + +/// Check whether a version matches a list of OR-ed constraints. +/// +/// * `version`: the effective version to test, or `None` if unknown. +/// * `constraints`: the list from `from_range` or `to_range`. +/// +/// Returns `true` if `version` matches at least one entry. +/// +/// Wildcard entries match any version including unknown. +/// Non-wildcard entries against an unknown version return `false`. +pub(crate) fn version_matches_range(version: Option<&Version>, constraints: &[String]) -> bool { + if constraints.is_empty() { + return false; + } + + for raw in constraints { + match parse_version_constraint(raw) { + Ok(VersionConstraint::Wildcard) => return true, + Ok(constraint) => { + if version.is_some_and(|ver| constraint.matches(ver)) { + return true; + } + // Unknown version + non-wildcard → doesn't match this entry + } + Err(e) => { + println!( + "Warning: invalid version constraint '{}' — skipping this entry: {e}", + raw + ); + } + } + } + + false +} + +/// Try to infer a version string by running `/greptime --version`. +/// Returns `None` if the binary cannot be executed or the output isn't parseable. +pub(crate) fn try_infer_version(bins_dir: &Path) -> Option { + let binary = bins_dir.join("greptime"); + if !binary.is_file() { + return None; + } + let output = Command::new(&binary).arg("--version").output().ok()?; + if !output.status.success() { + return None; + } + let stdout = String::from_utf8_lossy(&output.stdout); + // Typical output: "greptime 0.9.5-xxxxx" or just "greptime 0.9.5" + // Grab the first token that looks like a version. + for token in stdout.split_whitespace() { + // Strip leading 'v' if present and try to parse + let candidate = token.trim(); + let looks_like_version = candidate.starts_with('v') + || candidate.chars().next().is_some_and(|c| c.is_ascii_digit()); + let parsed_version = looks_like_version + .then(|| Version::parse(candidate).ok()) + .flatten(); + if let Some(ver) = parsed_version { + return Some(ver); + } + } + None +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_sanitize_namespace() { + assert_eq!(sanitize_namespace("basic_table"), "basic_table"); + assert_eq!(sanitize_namespace("my-case"), "my_case"); + assert_eq!(sanitize_namespace("123abc"), "c_123abc"); + assert_eq!(sanitize_namespace("UPPER"), "upper"); + assert_eq!(sanitize_namespace("a.b-c"), "a_b_c"); + } + + #[test] + fn test_validate_cases_metadata_rejects_empty_required_vectors() { + let case = CompatCase { + metadata: CaseMetadata { + name: "case".to_string(), + reason: "reason".to_string(), + introduced_by: "pr".to_string(), + topologies: vec![], + from_range: vec!["*".to_string()], + to_range: vec!["*".to_string()], + features: vec!["table".to_string()], + owner: "team".to_string(), + namespace: None, + }, + dir: PathBuf::from("case"), + namespace: "case".to_string(), + }; + + assert!(validate_cases_metadata(&[case]).is_err()); + } + + #[test] + fn test_validate_cases_metadata_catches_invalid_version_constraint() { + let case = CompatCase { + metadata: CaseMetadata { + name: "bad_constraint".to_string(), + reason: "test".to_string(), + introduced_by: "test".to_string(), + topologies: vec!["distributed".to_string()], + from_range: vec![">=not-a-version".to_string()], + to_range: vec!["*".to_string()], + features: vec!["table".to_string()], + owner: "test".to_string(), + namespace: None, + }, + dir: PathBuf::from("bad_constraint"), + namespace: "bad_constraint".to_string(), + }; + + assert!(validate_cases_metadata(&[case]).is_err()); + } + + #[test] + fn test_validate_case_namespaces_rejects_duplicate() { + let case_a = CompatCase { + metadata: CaseMetadata { + name: "case_a".to_string(), + reason: "test".to_string(), + introduced_by: "test".to_string(), + topologies: vec!["distributed".to_string()], + from_range: vec!["*".to_string()], + to_range: vec!["*".to_string()], + features: vec!["table".to_string()], + owner: "test".to_string(), + namespace: None, + }, + dir: PathBuf::from("case_a"), + namespace: "shared_name".to_string(), + }; + let case_b = CompatCase { + metadata: CaseMetadata { + name: "case_b".to_string(), + reason: "test".to_string(), + introduced_by: "test".to_string(), + topologies: vec!["distributed".to_string()], + from_range: vec!["*".to_string()], + to_range: vec!["*".to_string()], + features: vec!["table".to_string()], + owner: "test".to_string(), + namespace: None, + }, + dir: PathBuf::from("case_b"), + namespace: "shared_name".to_string(), + }; + + assert!(validate_cases_metadata(&[case_a.clone(), case_b.clone()]).is_ok()); + assert!(validate_case_namespaces(&[case_a, case_b]).is_err()); + } + + #[test] + fn test_validate_case_namespaces_rejects_any_duplicate() { + // All duplicate namespaces are rejected — the removed `isolation = "shared"` + // exemption no longer applies. + let case_a = CompatCase { + metadata: CaseMetadata { + name: "case_a".to_string(), + reason: "test".to_string(), + introduced_by: "test".to_string(), + topologies: vec!["distributed".to_string()], + from_range: vec!["*".to_string()], + to_range: vec!["*".to_string()], + features: vec!["table".to_string()], + owner: "test".to_string(), + namespace: Some("shared_name".to_string()), + }, + dir: PathBuf::from("case_a"), + namespace: "shared_name".to_string(), + }; + let case_b = CompatCase { + metadata: CaseMetadata { + name: "case_b".to_string(), + reason: "test".to_string(), + introduced_by: "test".to_string(), + topologies: vec!["distributed".to_string()], + from_range: vec!["*".to_string()], + to_range: vec!["*".to_string()], + features: vec!["table".to_string()], + owner: "test".to_string(), + namespace: Some("shared_name".to_string()), + }, + dir: PathBuf::from("case_b"), + namespace: "shared_name".to_string(), + }; + + assert!(validate_case_namespaces(&[case_a, case_b]).is_err()); + } + + #[test] + fn test_validate_cases_metadata_rejects_invalid_version_range() { + let case = CompatCase { + metadata: CaseMetadata { + name: "case".to_string(), + reason: "reason".to_string(), + introduced_by: "pr".to_string(), + topologies: vec!["distributed".to_string()], + from_range: vec![">=not-a-version".to_string()], + to_range: vec!["*".to_string()], + features: vec!["table".to_string()], + owner: "team".to_string(), + namespace: None, + }, + dir: PathBuf::from("case"), + namespace: "case".to_string(), + }; + + assert!(validate_cases_metadata(&[case]).is_err()); + } + + /// Write minimal required files for a compat case into `dir`. + fn write_minimal_case(dir: &Path) { + std::fs::create_dir_all(dir).unwrap(); + let case_toml = dir.join("case.toml"); + let setup_sql = dir.join("setup.sql"); + let verify_sql = dir.join("verify.sql"); + + std::fs::write( + &case_toml, + r#" +name = "test_case" +reason = "test" +introduced_by = "test" +topologies = ["distributed"] +from_range = ["*"] +to_range = ["*"] +features = ["table"] +owner = "test" +"#, + ) + .unwrap(); + std::fs::write(&setup_sql, "CREATE TABLE t (a INT);").unwrap(); + std::fs::write(&verify_sql, "SELECT * FROM t;").unwrap(); + } + + #[test] + fn test_discover_cases_allows_missing_verify_result() { + let tmp = tempfile::tempdir().unwrap(); + let case_dir = tmp.path().join("my_case"); + write_minimal_case(&case_dir); + // verify.result is intentionally absent — discovery should still succeed + assert!(!case_dir.join("verify.result").is_file()); + + let cases = + discover_cases(tmp.path()).expect("discover should succeed without verify.result"); + assert_eq!(cases.len(), 1); + assert_eq!(cases[0].metadata.name, "test_case"); + } + + #[test] + fn test_discover_cases_rejects_missing_setup_sql() { + let tmp = tempfile::tempdir().unwrap(); + let case_dir = tmp.path().join("my_case"); + write_minimal_case(&case_dir); + std::fs::remove_file(case_dir.join("setup.sql")).unwrap(); + + assert!(discover_cases(tmp.path()).is_err()); + } + + #[test] + fn test_discover_cases_rejects_missing_verify_sql() { + let tmp = tempfile::tempdir().unwrap(); + let case_dir = tmp.path().join("my_case"); + write_minimal_case(&case_dir); + std::fs::remove_file(case_dir.join("verify.sql")).unwrap(); + + assert!(discover_cases(tmp.path()).is_err()); + } + + #[test] + fn test_discover_cases_rejects_unknown_isolation_field() { + let tmp = tempfile::tempdir().unwrap(); + let case_dir = tmp.path().join("my_case"); + write_minimal_case(&case_dir); + std::fs::write( + case_dir.join("case.toml"), + r#" +name = "test_case" +reason = "test" +introduced_by = "test" +topologies = ["distributed"] +from_range = ["*"] +to_range = ["*"] +features = ["table"] +owner = "test" +isolation = "shared" +"#, + ) + .unwrap(); + + let err = discover_cases(tmp.path()).unwrap_err(); + assert!(err.contains("unknown field `isolation`")); + } + + #[test] + fn test_discover_cases_rejects_missing_case_toml() { + let tmp = tempfile::tempdir().unwrap(); + let case_dir = tmp.path().join("my_case"); + write_minimal_case(&case_dir); + std::fs::remove_file(case_dir.join("case.toml")).unwrap(); + + // No case.toml → skipped (not an error) + let result = discover_cases(tmp.path()); + assert!(result.is_err()); // no cases found at all + } + + // ------------------------------------------------------------------ + // Version-range matching tests + // ------------------------------------------------------------------ + + fn mkver(s: &str) -> Version { + Version::parse(s).unwrap() + } + + #[test] + fn test_parse_version_constraint_wildcard() { + let c = parse_version_constraint("*").unwrap(); + assert!(matches!(c, VersionConstraint::Wildcard)); + } + + #[test] + fn test_parse_version_constraint_exact() { + let c = parse_version_constraint("v1.2.3").unwrap(); + assert!(matches!(c, VersionConstraint::Exact(ref v) if v == &mkver("v1.2.3"))); + } + + #[test] + fn test_parse_version_constraint_gte() { + let c = parse_version_constraint(">=v1.1.0").unwrap(); + assert!(matches!(c, VersionConstraint::Gte(ref v) if v == &mkver("v1.1.0"))); + } + + #[test] + fn test_parse_version_constraint_lte() { + let c = parse_version_constraint("<=v1.1.0").unwrap(); + assert!(matches!(c, VersionConstraint::Lte(ref v) if v == &mkver("v1.1.0"))); + } + + #[test] + fn test_parse_version_constraint_gt() { + let c = parse_version_constraint(">v1.0.0").unwrap(); + assert!(matches!(c, VersionConstraint::Gt(ref v) if v == &mkver("v1.0.0"))); + } + + #[test] + fn test_parse_version_constraint_lt() { + let c = parse_version_constraint("=v1.1.1".to_string()]; + + // from matches <=v1.1.0 + assert!(version_matches_range(Some(&mkver("v0.9.5")), &from_range)); + assert!(version_matches_range(Some(&mkver("v1.1.0")), &from_range)); + assert!(!version_matches_range(Some(&mkver("v1.1.1")), &from_range)); + assert!(!version_matches_range(Some(&mkver("v1.2.0")), &from_range)); + + // to matches >=v1.1.1 + assert!(version_matches_range(Some(&mkver("v1.1.1")), &to_range)); + assert!(version_matches_range(Some(&mkver("v2.0.0")), &to_range)); + assert!(!version_matches_range(Some(&mkver("v1.1.0")), &to_range)); + assert!(!version_matches_range(Some(&mkver("v0.9.5")), &to_range)); + } + + #[test] + fn test_version_matches_range_unknown_version() { + // Non-wildcard ranges should NOT match unknown version + assert!(!version_matches_range(None, &["<=v1.1.0".to_string()])); + assert!(!version_matches_range(None, &[">=v1.1.1".to_string()])); + assert!(!version_matches_range(None, &["v1.0.0".to_string()])); + // Wildcard still matches unknown + assert!(version_matches_range(None, &["*".to_string()])); + } + + #[test] + fn test_version_matches_range_exact() { + let range = vec!["v1.0.0".to_string(), "v2.0.0".to_string()]; + assert!(version_matches_range(Some(&mkver("v1.0.0")), &range)); + assert!(version_matches_range(Some(&mkver("v2.0.0")), &range)); + assert!(!version_matches_range(Some(&mkver("v1.5.0")), &range)); + } + + #[test] + fn test_version_parse_without_v() { + let ver = Version::parse("1.2.3").unwrap(); + assert_eq!(ver.major, 1); + assert_eq!(ver.minor, 2); + assert_eq!(ver.patch, 3); + } + + #[test] + fn test_version_parse_with_suffix() { + let ver = Version::parse("1.2.3-alpha+build").unwrap(); + assert_eq!(ver.major, 1); + assert_eq!(ver.minor, 2); + assert_eq!(ver.patch, 3); + } + + #[test] + fn test_try_infer_version_no_binary() { + let tmp = tempfile::tempdir().unwrap(); + assert!(try_infer_version(tmp.path()).is_none()); + } +} diff --git a/tests/runner/src/env/bare.rs b/tests/runner/src/env/bare.rs index 61e93d26db..5bdd90a198 100644 --- a/tests/runner/src/env/bare.rs +++ b/tests/runner/src/env/bare.rs @@ -24,6 +24,7 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use async_trait::async_trait; +use common_error::ext::ErrorExt; use sqlness::{Database, EnvController, QueryContext}; use tokio::sync::Mutex as TokioMutex; @@ -42,6 +43,8 @@ const SERVER_MODE_METASRV_IDX: usize = 0; const SERVER_MODE_DATANODE_START_IDX: usize = 1; const SERVER_MODE_FRONTEND_IDX: usize = 4; const SERVER_MODE_FLOWNODE_IDX: usize = 5; +// Number of datanodes in distributed mode +const DISTRIBUTED_DATANODE_COUNT: usize = 3; #[derive(Clone)] pub enum WalConfig { @@ -174,6 +177,11 @@ impl Env { } async fn start_distributed(&self, id: usize) -> GreptimeDB { + self.start_distributed_inner(id).await + } + + /// Internal: start a distributed cluster with flownode. + async fn start_distributed_inner(&self, id: usize) -> GreptimeDB { if self.server_addrs.server_addr.is_some() { self.connect_db(&self.server_addrs, id).await } else { @@ -202,15 +210,13 @@ impl Env { db_ctx.set_server_mode(meta_server_mode.clone(), SERVER_MODE_METASRV_IDX); let meta_server = self.start_server(meta_server_mode, &db_ctx, id, true).await; - let datanode_1_mode = ServerMode::random_datanode(metasrv_port, 0); - db_ctx.set_server_mode(datanode_1_mode.clone(), SERVER_MODE_DATANODE_START_IDX); - let datanode_1 = self.start_server(datanode_1_mode, &db_ctx, id, true).await; - let datanode_2_mode = ServerMode::random_datanode(metasrv_port, 1); - db_ctx.set_server_mode(datanode_2_mode.clone(), SERVER_MODE_DATANODE_START_IDX + 1); - let datanode_2 = self.start_server(datanode_2_mode, &db_ctx, id, true).await; - let datanode_3_mode = ServerMode::random_datanode(metasrv_port, 2); - db_ctx.set_server_mode(datanode_3_mode.clone(), SERVER_MODE_DATANODE_START_IDX + 2); - let datanode_3 = self.start_server(datanode_3_mode, &db_ctx, id, true).await; + let mut datanodes = Vec::with_capacity(DISTRIBUTED_DATANODE_COUNT); + for i in 0..DISTRIBUTED_DATANODE_COUNT { + let datanode_mode = ServerMode::random_datanode(metasrv_port, i as u32); + db_ctx.set_server_mode(datanode_mode.clone(), SERVER_MODE_DATANODE_START_IDX + i); + let datanode = self.start_server(datanode_mode, &db_ctx, id, true).await; + datanodes.push(datanode); + } let frontend_mode = ServerMode::random_frontend(metasrv_port); let server_addr = frontend_mode.server_addr().unwrap(); @@ -224,9 +230,7 @@ impl Env { let mut greptimedb = self.connect_db(&server_addr, id).await; greptimedb.metasrv_process = Some(meta_server).into(); - greptimedb.server_processes = Some(Arc::new(Mutex::new(vec![ - datanode_1, datanode_2, datanode_3, - ]))); + greptimedb.server_processes = Some(Arc::new(Mutex::new(datanodes))); greptimedb.frontend_process = Some(frontend).into(); greptimedb.flownode_process = Some(flownode).into(); greptimedb.is_standalone = false; @@ -360,7 +364,7 @@ impl Env { } /// stop and restart the server process - async fn restart_server(&self, db: &GreptimeDB, is_full_restart: bool) { + pub(crate) async fn restart_server(&self, db: &GreptimeDB, is_full_restart: bool) { let bins_dir = db.active_bins_dir.lock().unwrap().clone().expect( "GreptimeDB binary is not available. Please pass in the path to the directory that contains the pre-built GreptimeDB binary. Or you may call `self.build_db()` beforehand.", ); @@ -386,6 +390,7 @@ impl Env { } } + // Stop flownode if present. if let Some(mut flownode_process) = db.flownode_process.lock().expect("poisoned lock").take() { @@ -441,7 +446,7 @@ impl Env { } let mut processes = vec![]; - for i in 0..3 { + for i in 0..DISTRIBUTED_DATANODE_COUNT { let datanode_mode = db .ctx .get_server_mode(SERVER_MODE_DATANODE_START_IDX + i) @@ -465,6 +470,7 @@ impl Env { .get_server_mode(SERVER_MODE_FRONTEND_IDX) .cloned() .unwrap(); + let server_addr = frontend_mode.server_addr().unwrap(); let frontend = self .start_server_with_bins_dir( frontend_mode, @@ -478,20 +484,35 @@ impl Env { .lock() .expect("lock poisoned") .replace(frontend); + + // Reconnect protocol clients to the new frontend process + // so that MySQL/Postgres queries use the restarted frontend, + // not stale connections to the old (killed) process. + let mut client = db.client.lock().await; + client + .reconnect_mysql_client(server_addr.mysql_server_addr.as_ref().unwrap()) + .await; + client + .reconnect_pg_client(server_addr.pg_server_addr.as_ref().unwrap()) + .await; } - let flownode_mode = db - .ctx - .get_server_mode(SERVER_MODE_FLOWNODE_IDX) - .cloned() - .unwrap(); - let flownode = self - .start_server_with_bins_dir(flownode_mode, &db.ctx, db.id, false, bins_dir.clone()) - .await; - db.flownode_process - .lock() - .expect("lock poisoned") - .replace(flownode); + // Restart flownode. + if let Some(flownode_mode) = db.ctx.get_server_mode(SERVER_MODE_FLOWNODE_IDX).cloned() { + let flownode = self + .start_server_with_bins_dir( + flownode_mode, + &db.ctx, + db.id, + false, + bins_dir.clone(), + ) + .await; + db.flownode_process + .lock() + .expect("lock poisoned") + .replace(flownode); + } processes }; @@ -587,6 +608,45 @@ impl Env { pub(crate) fn extra_args(&self) -> &Vec { &self.extra_args } + + /// Start a distributed GreptimeDB cluster. Exposed for compat runner. + pub(crate) async fn compat_start_distributed(&self, id: usize) -> GreptimeDB { + self.start_distributed(id).await + } + + /// Full restart of all distributed processes with a new binary directory, + /// preserving the same context and data. + /// After restart, waits for the frontend gRPC endpoint to become ready. + pub(crate) async fn compat_restart_all(&self, db: &GreptimeDB, bins_dir: PathBuf) { + *db.active_bins_dir.lock().unwrap() = Some(bins_dir); + self.restart_server(db, true).await; + self.wait_frontend_ready(db).await; + } + + /// Wait for frontend gRPC readiness after restart. + async fn wait_frontend_ready(&self, db: &GreptimeDB) { + let frontend_mode = db + .ctx + .get_server_mode(SERVER_MODE_FRONTEND_IDX) + .cloned() + .unwrap(); + if let Some(addr) = frontend_mode.check_addrs().first() { + println!("Waiting for frontend gRPC readiness at {addr}..."); + crate::util::retry_with_backoff( + || async { + let mut client = db.client.lock().await; + match client.grpc_query("SELECT 1").await { + Ok(_) => Ok(()), + Err(e) => Err(format!("Frontend not ready: {e}")), + } + }, + 10, + std::time::Duration::from_secs(1), + ) + .await + .unwrap_or_else(|e| panic!("Frontend failed to become ready: {e}")); + } + } } pub struct GreptimeDB { @@ -629,6 +689,81 @@ impl GreptimeDB { Err(e) => Box::new(ErrorFormatter::from(e)), } } + + /// Handle `QueryContext` directives for compat statement execution. + /// + /// Inspects `QueryContext` keys set by sqlness interceptors: + /// - `restart`: restarts the server (datanode-only) if not using external address. + /// - `version`: switches to the specified binary version and performs a full restart. + /// + /// This does **not** execute queries itself; it only prepares the server state. + /// Used by the compat runner. + pub(crate) async fn compat_prepare_query_context(&self, ctx: &QueryContext) { + if ctx.context.contains_key("restart") && self.env.server_addrs.server_addr.is_none() { + self.env.restart_server(self, false).await; + } else if let Some(version) = ctx.context.get("version") { + let version_bin_dir = self + .env + .versioned_bins_dirs + .lock() + .expect("lock poison") + .get(version.as_str()) + .cloned(); + + match version_bin_dir { + Some(path) if path.join(PROGRAM).is_file() => { + *self.active_bins_dir.lock().unwrap() = Some(path); + } + _ => { + maybe_pull_binary(version, self.env.pull_version_on_need).await; + let root = get_workspace_root(); + let new_path = PathBuf::from_iter([&root, version]); + *self.active_bins_dir.lock().unwrap() = Some(new_path); + } + } + + self.env.restart_server(self, true).await; + // sleep for a while to wait for the server to fully boot up + tokio::time::sleep(Duration::from_secs(5)).await; + } + } + + pub(crate) async fn compat_query( + &self, + query: &str, + ctx: &QueryContext, + ) -> Result { + let mut client = self.client.lock().await; + + // Handle protocol switching + if let Some(protocol) = ctx.context.get(PROTOCOL_KEY) { + if protocol == MYSQL { + return match client.mysql_query(query).await { + Ok(res) => Ok(crate::formatter::MysqlFormatter::from(res).to_string()), + Err(e) => Err(e), + }; + } else { + // postgres + return match client.postgres_query(query).await { + Ok(rows) => Ok(crate::formatter::PostgresqlFormatter::from(rows).to_string()), + Err(e) => Err(e), + }; + } + } + + // Default: gRPC + match client.grpc_query(query).await { + Ok(output) => Ok(OutputFormatter::from(output).to_string()), + Err(e) => { + let status_code = e.status_code(); + let root_cause = e.output_msg(); + Err(format!( + "Error: {}({status_code}), {root_cause}", + status_code as u32 + )) + } + } + } } #[async_trait] @@ -721,6 +856,11 @@ impl GreptimeDB { util::teardown_wal(); } } + + /// Stop all processes managed by this GreptimeDB. Exposed for compat runner. + pub(crate) fn compat_stop(&mut self) { + self.stop(); + } } impl Drop for GreptimeDB { diff --git a/tests/runner/src/main.rs b/tests/runner/src/main.rs index c6cf5dd4b5..ec41ca5671 100644 --- a/tests/runner/src/main.rs +++ b/tests/runner/src/main.rs @@ -32,6 +32,7 @@ async fn main() { match cmd.subcmd { SubCommand::Bare(cmd) => cmd.run().await, + SubCommand::Compat(cmd) => cmd.run().await, SubCommand::Kube(cmd) => cmd.run().await, } }