mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-07-06 22:10:39 +00:00
test: add sqlness compatibility runner (#8334)
* test: add sqlness compatibility runner Signed-off-by: discord9 <discord9@163.com> * fix: update remote dyn filter request Signed-off-by: discord9 <discord9@163.com> * fix: satisfy compat runner clippy Signed-off-by: discord9 <discord9@163.com> * test: add legacy jsonb compatibility case Signed-off-by: discord9 <discord9@163.com> * test: align legacy jsonb compat assertion Signed-off-by: discord9 <discord9@163.com> * fix: align remote dyn filter proto fields Signed-off-by: discord9 <discord9@163.com> * test: simplify compat topology and SQL handling Signed-off-by: discord9 <discord9@163.com> * fix: switch compat runner to target binary Signed-off-by: discord9 <discord9@163.com> * test: drop compat SQLNESS comment handling Signed-off-by: discord9 <discord9@163.com> * test: support sqlness commands in compat Signed-off-by: discord9 <discord9@163.com> * test: fix compat protocol reconnect Signed-off-by: discord9 <discord9@163.com> * test: filter compat cases by version Signed-off-by: discord9 <discord9@163.com> * test: add merge mode compat case Signed-off-by: discord9 <discord9@163.com> * test: harden compat runner setup Signed-off-by: discord9 <discord9@163.com> * test: address compat review follow-ups Signed-off-by: discord9 <discord9@163.com> * test: tighten compat case validation Signed-off-by: discord9 <discord9@163.com> * test: require explicit compat expectations Signed-off-by: discord9 <discord9@163.com> * test: generate missing compat snapshots Signed-off-by: discord9 <discord9@163.com> * test: add compat dry run Signed-off-by: discord9 <discord9@163.com> --------- Signed-off-by: discord9 <discord9@163.com>
This commit is contained in:
discord9
GitHub
2
Cargo.lock
generated
2
Cargo.lock
generated
@@ -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]]
|
||||
|
||||
53
tests/compatibility/AGENTS.md
Normal file
53
tests/compatibility/AGENTS.md
Normal file
@@ -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.<table>`. 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.
|
||||
148
tests/compatibility/README.md
Normal file
148
tests/compatibility/README.md
Normal file
@@ -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 <version>` to auto-pull a release, or `--from-bins-dir <path>` 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 <path>`.
|
||||
- **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. |
|
||||
| `"<vX.Y.Z"` | Matches versions strictly earlier than X.Y.Z. |
|
||||
|
||||
The range list is **OR**: a case matches if **any** entry matches.
|
||||
|
||||
**Best-effort enforcement**: The runner tries to determine the effective version:
|
||||
- `--from-version` is used directly.
|
||||
- `--from-bins-dir` / `--to-bins-dir` (or the default debug build) runs `<binary> --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.
|
||||
|
||||
```
|
||||
<statement>;
|
||||
|
||||
<output>
|
||||
|
||||
<next statement>;
|
||||
|
||||
<output>
|
||||
|
||||
```
|
||||
|
||||
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.<table>`.
|
||||
- **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 <ns>` via gRPC; then `USE <ns>` for gRPC/MySQL statements or `SET search_path TO '<ns>'` 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.
|
||||
8
tests/compatibility/cases/basic_table/case.toml
Normal file
8
tests/compatibility/cases/basic_table/case.toml
Normal file
@@ -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"
|
||||
13
tests/compatibility/cases/basic_table/setup.sql
Normal file
13
tests/compatibility/cases/basic_table/setup.sql
Normal file
@@ -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);
|
||||
12
tests/compatibility/cases/basic_table/verify.result
Normal file
12
tests/compatibility/cases/basic_table/verify.result
Normal file
@@ -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 |
|
||||
+---------------------+---+----------+-----+
|
||||
1
tests/compatibility/cases/basic_table/verify.sql
Normal file
1
tests/compatibility/cases/basic_table/verify.sql
Normal file
@@ -0,0 +1 @@
|
||||
SELECT ts, i, s, f FROM foo ORDER BY ts;
|
||||
8
tests/compatibility/cases/legacy_jsonb/case.toml
Normal file
8
tests/compatibility/cases/legacy_jsonb/case.toml
Normal file
@@ -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"
|
||||
12
tests/compatibility/cases/legacy_jsonb/setup.sql
Normal file
12
tests/compatibility/cases/legacy_jsonb/setup.sql
Normal file
@@ -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');
|
||||
39
tests/compatibility/cases/legacy_jsonb/verify.result
Normal file
39
tests/compatibility/cases/legacy_jsonb/verify.result
Normal file
@@ -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 |
|
||||
| | |
|
||||
+--------------+---------------------------------------------+
|
||||
10
tests/compatibility/cases/legacy_jsonb/verify.sql
Normal file
10
tests/compatibility/cases/legacy_jsonb/verify.sql
Normal file
@@ -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;
|
||||
8
tests/compatibility/cases/merge_mode/case.toml
Normal file
8
tests/compatibility/cases/merge_mode/case.toml
Normal file
@@ -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"
|
||||
19
tests/compatibility/cases/merge_mode/setup.sql
Normal file
19
tests/compatibility/cases/merge_mode/setup.sql
Normal file
@@ -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');
|
||||
41
tests/compatibility/cases/merge_mode/verify.result
Normal file
41
tests/compatibility/cases/merge_mode/verify.result
Normal file
@@ -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' |
|
||||
| | ) |
|
||||
+------------+-------------------------------------------+
|
||||
10
tests/compatibility/cases/merge_mode/verify.sql
Normal file
10
tests/compatibility/cases/merge_mode/verify.sql
Normal file
@@ -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;
|
||||
@@ -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 }}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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),
|
||||
}
|
||||
|
||||
|
||||
878
tests/runner/src/cmd/compat.rs
Normal file
878
tests/runner/src/cmd/compat.rs
Normal file
@@ -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<String>,
|
||||
|
||||
/// Path to the directory containing the "from" GreptimeDB binary.
|
||||
#[clap(long)]
|
||||
from_bins_dir: Option<PathBuf>,
|
||||
|
||||
/// Path to the directory containing the "to" GreptimeDB binary.
|
||||
/// Defaults to the current debug build.
|
||||
#[clap(long)]
|
||||
to_bins_dir: Option<PathBuf>,
|
||||
|
||||
/// Directory of compatibility test cases.
|
||||
/// Defaults to `tests/compatibility/cases` relative to workspace root.
|
||||
#[clap(long)]
|
||||
case_dir: Option<PathBuf>,
|
||||
|
||||
/// 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 `<dir>/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 <ns>` 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<String>,
|
||||
display_query: Vec<String>,
|
||||
execute_query: Vec<String>,
|
||||
interceptors: Vec<InterceptorRef>,
|
||||
}
|
||||
|
||||
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>), 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<Vec<ParsedStatement>, 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");
|
||||
}
|
||||
}
|
||||
822
tests/runner/src/cmd/compat_case.rs
Normal file
822
tests/runner/src/cmd/compat_case.rs
Normal file
@@ -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<String>,
|
||||
/// Version range for the "from" binary. `*` means all versions.
|
||||
pub from_range: Vec<String>,
|
||||
/// Version range for the "to" binary. `*` means all versions.
|
||||
pub to_range: Vec<String>,
|
||||
/// Features required (e.g. ["table", "flow"]).
|
||||
pub features: Vec<String>,
|
||||
/// 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<String>,
|
||||
}
|
||||
|
||||
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<Vec<CompatCase>, 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<Self, String> {
|
||||
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::<u64>()
|
||||
.map_err(|_| format!("Invalid major version in '{}'", raw))?;
|
||||
let minor = parts[1]
|
||||
.parse::<u64>()
|
||||
.map_err(|_| format!("Invalid minor version in '{}'", raw))?;
|
||||
let patch = parts[2]
|
||||
.parse::<u64>()
|
||||
.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<VersionConstraint, String> {
|
||||
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 `<bins_dir>/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<Version> {
|
||||
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("<v2.0.0").unwrap();
|
||||
assert!(matches!(c, VersionConstraint::Lt(ref v) if v == &mkver("v2.0.0")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_parse_version_constraint_eq_double() {
|
||||
let c = parse_version_constraint("==v1.0.0").unwrap();
|
||||
assert!(matches!(c, VersionConstraint::Exact(ref v) if v == &mkver("v1.0.0")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_version_matches_range_wildcard() {
|
||||
assert!(version_matches_range(None, &["*".to_string()]));
|
||||
assert!(version_matches_range(
|
||||
Some(&mkver("v9.9.9")),
|
||||
&["*".to_string()]
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_version_matches_range_legacy_jsonb() {
|
||||
let from_range = vec!["<=v1.1.0".to_string()];
|
||||
let to_range = vec![">=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());
|
||||
}
|
||||
}
|
||||
192
tests/runner/src/env/bare.rs
vendored
192
tests/runner/src/env/bare.rs
vendored
@@ -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<String> {
|
||||
&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<String, String> {
|
||||
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 {
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user