diff --git a/.github/scripts/metric-snapshot-fixture.py b/.github/scripts/metric-snapshot-fixture.py
new file mode 100644
index 00000000000..4341bc6f579
--- /dev/null
+++ b/.github/scripts/metric-snapshot-fixture.py
@@ -0,0 +1,294 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Deterministic wide, heterogeneous Metric tables for snapshot benchmarks."""
+
+import argparse
+import base64
+import concurrent.futures
+import hashlib
+import http.client
+import json
+import os
+from pathlib import Path
+import threading
+import time
+import urllib.parse
+
+GROUPS = 10
+TAG_POOL = 298
+ROWS = 32
+DATABASE = "metrics"
+
+
+class Client:
+    def __init__(self, addr):
+        self.addr = addr
+        self.connections = threading.local()
+        self.headers = {"Content-Type": "application/x-www-form-urlencoded"}
+        if auth := os.environ.get("GREPTIME_BENCH_AUTH"):
+            self.headers["Authorization"] = "Basic " + base64.b64encode(
+                auth.encode()
+            ).decode()
+
+    def sql(self, query, database=DATABASE, batch=False):
+        connection = getattr(self.connections, "connection", None)
+        if connection is None:
+            connection = http.client.HTTPConnection(self.addr, timeout=600)
+            self.connections.connection = connection
+        body = urllib.parse.urlencode({"db": database, "sql": query})
+        try:
+            connection.request(
+                "POST", "/v1/ddl/logical-tables" if batch else "/v1/sql",
+                body, self.headers,
+            )
+            response = connection.getresponse()
+            raw = response.read()
+        except (OSError, http.client.HTTPException):
+            connection.close()
+            self.connections.connection = None
+            raise
+        data = json.loads(raw)
+        if response.status != 200 or "error" in data:
+            raise RuntimeError(f"SQL failed (HTTP {response.status}): {data}")
+        return data["output"]
+
+    def values(self, query, database=DATABASE):
+        return self.sql(query, database)[0]["records"]["rows"]
+
+
+def tags(index, per_group):
+    return sorted((index % per_group * 7 + offset) % TAG_POOL for offset in range(10))
+
+
+def column_names(tag_ids):
+    return ["ts", *[f"tag_{tag:03d}" for tag in tag_ids], "val"]
+
+
+def columns(tag_ids):
+    names = [f"tag_{tag:03d}" for tag in tag_ids]
+    return (
+        "ts TIMESTAMP TIME INDEX, "
+        + ", ".join(name + " STRING" for name in names)
+        + ", val DOUBLE, PRIMARY KEY(" + ",".join(names) + ")"
+    )
+
+
+def table_ddl(index, per_group):
+    return (
+        f"CREATE TABLE m{index:06d} ({columns(tags(index, per_group))}) "
+        f"ENGINE=metric WITH(on_physical_table='phy_{index // per_group:02d}')"
+    )
+
+
+def row(index, row_index, per_group):
+    values = []
+    for tag in tags(index, per_group):
+        if (row_index + tag) % 29 == 0:
+            value = None
+        elif (row_index + tag) % 17 == 0:
+            value = ""
+        elif tag % 10 == 0:
+            value = f"instance-{index:06d}-{row_index % 4}"
+        elif tag % 10 == 1:
+            value = "区域-" + str(row_index % 3)
+        elif tag % 10 == 2:
+            value = "label-" + "x" * 48 + "-" + str(row_index % 2)
+        else:
+            value = f"tag{tag:03d}-v{row_index % (2 + tag % 7)}"
+        values.append(value)
+    return [row_index, *values, None if row_index % 7 == 0 else index % 97 + row_index / 4]
+
+
+def sql_value(value):
+    if value is None:
+        return "NULL"
+    if isinstance(value, str):
+        return "'" + value.replace("'", "''") + "'"
+    return str(value)
+
+
+def canonical(value):
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+
+
+def update_digest(digest, value):
+    digest.update((canonical(value) + "\n").encode())
+
+
+def progress(phase, completed, total):
+    if completed == total or completed % 4096 == 0:
+        print(f"{phase}: {completed}/{total}", flush=True)
+
+
+def prepare(client, per_group):
+    stages = {}
+    started = time.monotonic()
+    client.sql(f"CREATE DATABASE {DATABASE}", "public")
+    for group in range(GROUPS):
+        client.sql(
+            f"CREATE TABLE phy_{group:02d} ({columns(range(TAG_POOL))}) "
+            "ENGINE=metric WITH(physical_metric_table='')"
+        )
+    total = GROUPS * per_group
+    for group in range(GROUPS):
+        for base in range(group * per_group, (group + 1) * per_group, 128):
+            end = min((group + 1) * per_group, base + 128)
+            client.sql(";".join(table_ddl(i, per_group) for i in range(base, end)), batch=True)
+            progress("create", end, total)
+    stages["create_seconds"] = time.monotonic() - started
+
+    def insert(index):
+        rows = ",".join(
+            "(" + ",".join(sql_value(v) for v in row(index, j, per_group)) + ")"
+            for j in range(ROWS)
+        )
+        names = ",".join(column_names(tags(index, per_group)))
+        client.sql(f"INSERT INTO m{index:06d} ({names}) VALUES {rows}")
+
+    started = time.monotonic()
+    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
+        for base in range(0, total, 128):
+            end = min(total, base + 128)
+            list(pool.map(insert, range(base, end)))
+            progress("insert", end, total)
+    stages["insert_seconds"] = time.monotonic() - started
+    started = time.monotonic()
+    for group in range(GROUPS):
+        client.sql(f"ADMIN FLUSH_TABLE('{DATABASE}.phy_{group:02d}')")
+    stages["flush_seconds"] = time.monotonic() - started
+    return stages
+
+
+def check_records(index, per_group, records):
+    names = column_names(tags(index, per_group))
+    types = ["TimestampMillisecond", *["String"] * 10, "Float64"]
+    expected_columns = [dict(name=name, data_type=kind) for name, kind in zip(names, types)]
+    actual_columns = records["schema"]["column_schemas"]
+    if sorted(actual_columns, key=lambda c: c["name"]) != sorted(
+        expected_columns, key=lambda c: c["name"]
+    ):
+        raise RuntimeError(f"logical schema mismatch: m{index:06d}")
+    # Metric tables canonicalize column order; SELECT * still checks every column.
+    positions = [names.index(column["name"]) for column in actual_columns]
+    expected_rows = [
+        [values[position] for position in positions]
+        for values in (row(index, j, per_group) for j in range(ROWS))
+    ]
+    # Canonical JSON also distinguishes integers, floats, booleans, and NULL.
+    if canonical(records["rows"]) != canonical(expected_rows):
+        raise RuntimeError(f"typed values/NULL mismatch: m{index:06d}")
+
+
+def validate(client, per_group):
+    total = GROUPS * per_group
+    expected_names = {f"m{i:06d}" for i in range(total)} | {
+        f"phy_{group:02d}" for group in range(GROUPS)
+    }
+    table_rows = client.values(
+        "SELECT table_name, table_id FROM information_schema.tables "
+        f"WHERE table_schema='{DATABASE}' ORDER BY table_name"
+    )
+    table_ids = dict(table_rows)
+    if len(table_rows) != len(expected_names) or set(table_ids) != expected_names:
+        raise RuntimeError("table membership mismatch")
+    if len(set(table_ids.values())) != len(table_ids):
+        raise RuntimeError("duplicate table IDs")
+    digests = {name: hashlib.sha256() for name in ("schema", "data", "ddl")}
+
+    def check_ddl(names):
+        outputs = client.sql(";".join(f"SHOW CREATE TABLE {name}" for name in names))
+        if len(outputs) != len(names):
+            raise RuntimeError("incomplete SHOW CREATE TABLE response")
+        for name, output in zip(names, outputs):
+            records = output["records"]["rows"]
+            if len(records) != 1 or records[0][0] != name or not records[0][1]:
+                raise RuntimeError(f"invalid SHOW CREATE TABLE response: {name}")
+            update_digest(digests["ddl"], records)
+
+    for base in range(0, total, 64):
+        end = min(total, base + 64)
+        names = [f"m{i:06d}" for i in range(base, end)]
+        outputs = client.sql(";".join(f"SELECT * FROM {name} ORDER BY ts" for name in names))
+        if len(outputs) != len(names):
+            raise RuntimeError("incomplete SELECT response")
+        for index, output in enumerate(outputs, base):
+            records = output["records"]
+            check_records(index, per_group, records)
+            update_digest(digests["schema"], [f"m{index:06d}", records["schema"]])
+            update_digest(digests["data"], [f"m{index:06d}", records["rows"]])
+        check_ddl(names)
+        progress("validate", end, total)
+    check_ddl([f"phy_{group:02d}" for group in range(GROUPS)])
+    return {**{name + "_sha256": digest.hexdigest() for name, digest in digests.items()},
+            "table_ids": table_ids}
+
+
+def compare_source(result, source):
+    for field in ("fixture", "schema_sha256", "data_sha256", "ddl_sha256"):
+        if result[field] != source[field]:
+            raise RuntimeError(f"source/target mismatch: {field}")
+    source_ids, target_ids = source["table_ids"], result["table_ids"]
+    if source_ids.keys() != target_ids.keys():
+        raise RuntimeError("source/target table membership mismatch")
+    changed = sum(source_ids[name] != target_ids[name] for name in source_ids)
+    if changed != len(source_ids):
+        raise RuntimeError(f"table IDs not all remapped: {changed}/{len(source_ids)}")
+    result["remapped_table_ids"] = changed
+
+
+def placeholders(client):
+    database = "metric_bench_guard"
+    client.sql(f"CREATE DATABASE {database}", "public")
+    for index in range(17):
+        client.sql(f"CREATE TABLE placeholder_{index} (ts TIMESTAMP TIME INDEX)", database)
+    return {"placeholder_tables": 17}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("action", choices=("prepare", "validate", "placeholders"))
+    parser.add_argument("--addr", default="127.0.0.1:4000")
+    parser.add_argument("--tables-per-group", type=int, default=10000)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--source", type=Path, help="Source validation JSON, for target comparison")
+    args = parser.parse_args()
+    if not 50 <= args.tables_per_group <= 10000:
+        parser.error("--tables-per-group must be between 50 and 10000")
+    if args.source is not None and args.action != "validate":
+        parser.error("--source only applies to validate")
+    coverage = {tag for i in range(args.tables_per_group) for tag in tags(i, args.tables_per_group)}
+    if coverage != set(range(TAG_POOL)):
+        parser.error("fixture does not cover all 298 tags")
+    client = Client(args.addr)
+    started = time.monotonic()
+    result = {"fixture": {"groups": GROUPS, "tables_per_group": args.tables_per_group,
+                          "rows_per_table": ROWS, "tag_pool": TAG_POOL}}
+    if args.action == "placeholders":
+        result.update(placeholders(client))
+    elif args.action == "prepare":
+        result.update(prepare(client, args.tables_per_group))
+    else:
+        result.update(validate(client, args.tables_per_group))
+    result["wall_seconds"] = time.monotonic() - started
+    if args.source is not None:
+        compare_source(result, json.loads(args.source.read_text()))
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n")
+    print(json.dumps({key: value for key, value in result.items() if key != "table_ids"}))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/.github/scripts/metric-snapshot-s3-bench.py b/.github/scripts/metric-snapshot-s3-bench.py
new file mode 100644
index 00000000000..7bb01a410b1
--- /dev/null
+++ b/.github/scripts/metric-snapshot-s3-bench.py
@@ -0,0 +1,392 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Run one packed Metric snapshot smoke, then one full S3 round trip."""
+
+import argparse
+from datetime import datetime
+import json
+import os
+from pathlib import Path
+import re
+import signal
+import subprocess
+import sys
+import threading
+import time
+import urllib.request
+
+HERE = Path(__file__).resolve().parent
+
+
+def command(args, **kwargs):
+    result = subprocess.run(args, text=True, stdout=subprocess.PIPE,
+                            stderr=subprocess.STDOUT, **kwargs)
+    if result.returncode:
+        # CLI argv and error output can contain storage credentials.
+        raise RuntimeError(f"{args[0]} exited with status {result.returncode}")
+    return result.stdout
+
+
+def write_json(path, value):
+    path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n")
+
+
+def metric_values(text):
+    values = {}
+    for line in text.splitlines():
+        if line and not line.startswith("#"):
+            key, value = line.rsplit(" ", 1)
+            values[key] = float(value)
+    return values
+
+
+def docker_timestamp(value):
+    # Docker emits nanoseconds; Python 3.9 accepts at most microseconds.
+    return datetime.fromisoformat(re.sub(r"(\.\d{6})\d+", r"\1", value).replace("Z", "+00:00"))
+
+
+def host_memory():
+    if not Path("/proc/vmstat").exists():
+        return {"available": False}
+    vm = dict(line.split() for line in Path("/proc/vmstat").read_text().splitlines())
+    mem = dict(line.split(":", 1) for line in Path("/proc/meminfo").read_text().splitlines())
+    return {"available": True, "pswpin_pages": int(vm["pswpin"]),
+            "pswpout_pages": int(vm["pswpout"]), "oom_kill": int(vm.get("oom_kill", 0)),
+            **{key + "_kib": int(mem[key].split()[0])
+               for key in ("MemTotal", "MemAvailable", "SwapTotal", "SwapFree")}}
+
+
+def process_usage(pid):
+    try:
+        stat = Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()
+        return {"cpu_seconds": (int(stat[11]) + int(stat[12])) / os.sysconf("SC_CLK_TCK"),
+                "rss_bytes": int(stat[21]) * os.sysconf("SC_PAGE_SIZE")}
+    except (OSError, IndexError):
+        return None
+
+
+class Benchmark:
+    def __init__(self, local_smoke):
+        self.local_smoke = local_smoke
+        self.image = os.environ["GREPTIMEDB_IMAGE"]
+        self.commit = os.environ["EXPECTED_COMMIT"]
+        if not re.fullmatch(r"greptime/greptimedb(?:-nightly)?@sha256:[0-9a-f]{64}", self.image):
+            raise ValueError("GREPTIMEDB_IMAGE must be an immutable official image digest")
+        if not re.fullmatch(r"[0-9a-f]{40}", self.commit):
+            raise ValueError("EXPECTED_COMMIT must be a full commit SHA")
+        self.per_group = int(os.environ.get("LOGICAL_TABLES_PER_GROUP", "10000"))
+        if not 50 <= self.per_group <= 10000:
+            raise ValueError("LOGICAL_TABLES_PER_GROUP must be in 50..10000")
+        self.prefix = os.environ["RUN_PREFIX"]
+        if not re.fullmatch(r"metric-snapshot-bench/[0-9]+-[0-9]+/", self.prefix):
+            raise ValueError("RUN_PREFIX must be metric-snapshot-bench/<run>-<attempt>/")
+        self.results = Path(os.environ["RESULTS"]).resolve()
+        self.data = Path(os.environ["DATA_ROOT"]).resolve()
+        self.results.mkdir(parents=True, exist_ok=False)
+        self.data.mkdir(parents=True, exist_ok=False)
+        self.network = "metric-snapshot-" + self.prefix.split("/")[1]
+        self.server = self.network + "-db"
+        self.cli_name = self.network + "-cli"
+        self.port = int(os.environ.get("BENCH_HTTP_PORT", "19410"))
+        self.base_url = f"http://127.0.0.1:{self.port}"
+        self.endpoint = os.environ.get("S3_ENDPOINT", "")
+        if self.endpoint and not local_smoke:
+            raise ValueError("Custom S3 endpoints are restricted to local smoke")
+        if not local_smoke and os.environ["S3_REGION"] != "ap-southeast-1":
+            raise ValueError("Cloud benchmark requires ap-southeast-1")
+        self.containers = set()
+        self.network_created = False
+        self.phases = []
+        self.server_pid = 0
+        self.round_dir = None
+
+    def docker(self, *args):
+        return command(["docker", *args])
+
+    def metrics(self):
+        try:
+            with urllib.request.urlopen(self.base_url + "/metrics", timeout=5) as response:
+                return response.read().decode()
+        except OSError:
+            return ""
+
+    def phase(self, name, action):
+        print(f"Starting {self.round_dir.name}/{name}", flush=True)
+        before = host_memory()
+        metrics_before = self.metrics()
+        samples = []
+        done = threading.Event()
+
+        def sample():
+            while not done.is_set():
+                samples.append({"time": time.time(), "server": process_usage(self.server_pid),
+                                "cli": process_usage(getattr(self, "cli_pid", 0)),
+                                "query_memory": {k: v for k, v in metric_values(self.metrics()).items()
+                                                 if "query_memory" in k} if self.server_pid else {}})
+                done.wait(0.5)
+
+        monitor = threading.Thread(target=sample, daemon=True)
+        monitor.start()
+        start = time.monotonic()
+        outcome = "failed"
+        result = None
+        try:
+            result = action()
+            outcome = "passed"
+            return result
+        finally:
+            elapsed = time.monotonic() - start
+            done.set()
+            monitor.join()
+            after = host_memory()
+            interfered = before["available"] and after["available"] and (
+                after["pswpout_pages"] > before["pswpout_pages"] or after["oom_kill"] > before["oom_kill"])
+            if interfered:
+                outcome = "resource-invalid"
+            record = {"round": self.round_dir.name, "phase": name, "outcome": outcome,
+                      "wall_seconds": elapsed, "host_before": before, "host_after": after,
+                      "process_samples": samples}
+            if isinstance(result, dict):
+                record.update(result)
+            if before["available"] and after["available"]:
+                record["new_swapout_pages"] = after["pswpout_pages"] - before["pswpout_pages"]
+                record["new_oom_kills"] = after["oom_kill"] - before["oom_kill"]
+            metrics_after = self.metrics()
+            pre, post = metric_values(metrics_before), metric_values(metrics_after)
+            record["server_counter_deltas"] = {
+                key: value - pre.get(key, 0) for key, value in post.items()
+                if ("opendal" in key and ("_count{" in key or "_sum{" in key))
+                or key.startswith("greptime_servers_http_requests_total{")}
+            write_json(self.round_dir / f"{name}-resources.json", record)
+            (self.round_dir / f"{name}-before.prom").write_text(metrics_before)
+            (self.round_dir / f"{name}-after.prom").write_text(metrics_after)
+            self.phases.append({k: v for k, v in record.items() if k != "process_samples"})
+            self.summary()
+            print(f"Finished {name}: {outcome}, {elapsed:.3f}s", flush=True)
+            if interfered:
+                raise RuntimeError(f"Resource interference during {name}; evidence retained")
+
+    def fixture(self, operation, per_group, output, source=None):
+        args = [sys.executable, str(HERE / "metric-snapshot-fixture.py"), operation,
+                "--addr", f"127.0.0.1:{self.port}", "--tables-per-group", str(per_group),
+                "--output", str(output)]
+        if source:
+            args += ["--source", str(source)]
+        with (self.round_dir / f"{operation}-{output.stem}.log").open("w") as log:
+            status = subprocess.run(args, stdout=log, stderr=subprocess.STDOUT).returncode
+        if status:
+            raise RuntimeError(f"Fixture {operation} failed; see its artifact log")
+
+    def start_server(self, role):
+        root = self.data / self.round_dir.name / role
+        root.mkdir(parents=True)
+        evidence = self.round_dir / role
+        evidence.mkdir()
+        config = root / "config.toml"
+        config.write_text('''enable_telemetry = false
+[http]
+addr = "0.0.0.0:4000"
+[grpc]
+bind_addr = "0.0.0.0:4001"
+[mysql]
+enable = false
+[postgres]
+enable = false
+[storage]
+type = "File"
+data_home = "/data/db"
+[query]
+parallelism = 4
+memory_pool_size = "2GiB"
+[logging]
+dir = "/evidence/logs"
+level = "info"
+''')
+        self.containers.add(self.server)
+        self.docker("run", "-d", "--name", self.server, "--network", self.network,
+                    "--network-alias", "metric-db", "--user", f"{os.getuid()}:{os.getgid()}",
+                    "--add-host", "host.docker.internal:host-gateway",
+                    "-p", f"127.0.0.1:{self.port}:4000",
+                    "-v", f"{root}:/data", "-v", f"{evidence}:/evidence",
+                    self.image, "standalone", "start", "-c", "/data/config.toml")
+        self.server_pid = int(self.docker("inspect", "-f", "{{.State.Pid}}", self.server))
+        deadline = time.monotonic() + 120
+        while time.monotonic() < deadline:
+            try:
+                with urllib.request.urlopen(self.base_url + "/v1/capabilities", timeout=2) as response:
+                    caps = json.load(response)
+                if any(caps.get(key) != 1 for key in
+                       ("metric_packed_export", "metric_packed_import", "metric_batch_ddl")):
+                    raise RuntimeError("Required packed snapshot capabilities are absent")
+                write_json(evidence / "capabilities.json", caps)
+                return
+            except OSError:
+                if self.docker("inspect", "-f", "{{.State.Running}}", self.server).strip() != "true":
+                    break
+                time.sleep(1)
+        raise RuntimeError("Server did not become ready")
+
+    def stop_server(self, role):
+        self.docker("stop", "-t", "60", self.server)
+        state = json.loads(self.docker("inspect", "-f", "{{json .State}}", self.server))
+        write_json(self.round_dir / role / "exit.json", state)
+        (self.round_dir / role / "stdout.log").write_text(self.docker("logs", self.server))
+        self.docker("rm", self.server)
+        self.containers.remove(self.server)
+        self.server_pid = 0
+        if state["OOMKilled"] or state["ExitCode"] != 0:
+            raise RuntimeError(f"{role} server did not shut down cleanly")
+
+    def cli(self, args, name):
+        # Expand credentials only inside the container. Never persist argv or env.
+        script = 'exec greptime "$@" --s3 --s3-region "$S3_REGION" --s3-access-key-id "$AWS_ACCESS_KEY_ID" --s3-secret-access-key "$AWS_SECRET_ACCESS_KEY"'
+        if self.endpoint:
+            args += ["--s3-endpoint", self.endpoint]
+        self.containers.add(self.cli_name)
+        self.docker("create", "--name", self.cli_name, "--network", self.network,
+                    "--user", f"{os.getuid()}:{os.getgid()}",
+                    "--add-host", "host.docker.internal:host-gateway",
+                    "-e", "AWS_ACCESS_KEY_ID", "-e", "AWS_SECRET_ACCESS_KEY", "-e", "S3_REGION",
+                    "-v", f"{self.round_dir}:/evidence", "--entrypoint", "/bin/sh", self.image,
+                    "-c", script, "sh", "--log-dir", f"/evidence/{name}-logs", "cli", "data", *args)
+
+        def execute():
+            self.docker("start", self.cli_name)
+            self.cli_pid = int(self.docker("inspect", "-f", "{{.State.Pid}}", self.cli_name))
+            status = int(self.docker("wait", self.cli_name))
+            state = json.loads(self.docker("inspect", "-f", "{{json .State}}", self.cli_name))
+            (self.round_dir / f"{name}.log").write_text(self.docker("logs", self.cli_name))
+            write_json(self.round_dir / f"{name}-exit.json", state)
+            if status or state["ExitCode"] or state["OOMKilled"]:
+                raise RuntimeError(f"{name} failed; see artifact log")
+            return {"cli_wall_seconds": (docker_timestamp(state["FinishedAt"]) -
+                                          docker_timestamp(state["StartedAt"])).total_seconds()}
+
+        try:
+            self.phase(name, execute)
+        finally:
+            self.cli_pid = 0
+            if not (self.round_dir / f"{name}-exit.json").exists():
+                self.save_container(self.cli_name, name + "-failure")
+            self.docker("rm", "-f", self.cli_name)
+            self.containers.remove(self.cli_name)
+
+    def round(self, label, per_group):
+        self.round_dir = self.results / label
+        self.round_dir.mkdir()
+        self.start_server("source")
+        source = self.round_dir / "source-validation.json"
+        self.phase("prepare", lambda: self.fixture("prepare", per_group, self.round_dir / "prepare.json"))
+        self.phase("source-validation", lambda: self.fixture("validate", per_group, source))
+        location = f"s3://{os.environ['S3_BUCKET']}/{self.prefix}{label}"
+        self.cli(["export-v2", "create", "--addr", "metric-db:4000", "--to", location,
+                  "--schemas", "metrics", "--pack-metric-data", "--parallelism", "4",
+                  "--chunk-parallelism", "1", "--start-time", "1970-01-01T00:00:00Z",
+                  "--end-time", "1970-01-01T00:00:01Z", "--timeout", "1h",
+                  "--no-proxy", "--progress", "always"], "export")
+        if not self.local_smoke:
+            command(["uv", "run", str(HERE / "metric-snapshot-s3-ops.py"), "inventory", "--round", label,
+                     "--output", str(self.round_dir / "s3-inventory.json")])
+        self.stop_server("source")
+        self.cli(["export-v2", "verify", "--snapshot", location], "verify")
+        self.start_server("target")
+        self.fixture("placeholders", per_group, self.round_dir / "placeholders.json")
+        self.cli(["import-v2", "--addr", "metric-db:4000", "--from", location,
+                  "--task-parallelism", "1", "--state-path", "/evidence/import-state.json",
+                  "--timeout", "1h", "--no-proxy", "--progress", "always"], "import")
+        imported = json.loads((self.round_dir / "import-resources.json").read_text())
+        batch_requests = sum(value for key, value in imported["server_counter_deltas"].items()
+                             if '/v1/ddl/logical-tables' in key and 'code="200"' in key)
+        if batch_requests <= 0:
+            raise RuntimeError("Import did not complete any successful batch DDL HTTP requests")
+        write_json(self.round_dir / "import-completion.json", {"batch_ddl_requests": batch_requests,
+                   "resume_state_removed": not (self.round_dir / "import-state.json").exists()})
+        if (self.round_dir / "import-state.json").exists():
+            raise RuntimeError("Successful import left its resume state file")
+        self.phase("target-validation", lambda: self.fixture(
+            "validate", per_group, self.round_dir / "target-validation.json", source))
+        self.stop_server("target")
+
+    def summary(self):
+        lines = ["# Metric packed snapshot benchmark", "", f"Image: `{self.image}`",
+                 f"Source commit: `{self.commit}`", "",
+                 "Local smoke only; timings are not AWS results." if self.local_smoke else
+                 "AWS S3 in ap-southeast-1; source and target use local File storage.", "",
+                 "| Round | Phase | Seconds | Result |", "| --- | --- | ---: | --- |"]
+        for phase in self.phases:
+            lines.append(f"| {phase['round']} | {phase['phase']} | {phase['wall_seconds']:.3f} | {phase['outcome']} |")
+        lines += ["", "Each logical table has 32 rows, 10 overlapping string tags, a timestamp and a DOUBLE.",
+                  "Phase durations include Docker start, wait and log collection. cli_wall_seconds records Docker StartedAt to FinishedAt separately; setup, loading and full validation are separate.",
+                  "Process CPU/RSS are sampled every 0.5 s on Linux, not exact process peaks. Linux host swap/OOM counters are checked per phase; configured swap is not RAM.",
+                  "Raw server Prometheus snapshots and CLI progress/logs are retained. OpenDAL operation counts are not AWS HTTP request counts: multipart requests, retries and CLI operations are not fully covered. Exact AWS request totals are unavailable."]
+        (self.results / "summary.md").write_text("\n".join(lines) + "\n")
+
+    def run(self):
+        self.docker("pull", self.image)
+        version = self.docker("run", "--rm", self.image, "--version")
+        (self.results / "version.txt").write_text(version)
+        if self.commit not in version or "clean: true" not in version:
+            raise RuntimeError("Image version does not match the expected clean source commit")
+        for args, required in ((["export-v2", "create"], "--pack-metric-data"),
+                               (["import-v2"], "--task-parallelism")):
+            help_text = self.docker("run", "--rm", self.image, "cli", "data", *args, "--help")
+            (self.results / (args[0] + "-help.txt")).write_text(help_text)
+            if required not in help_text:
+                raise RuntimeError(f"Image CLI lacks {required}")
+        write_json(self.results / "host.json", {"memory": host_memory(), "local_smoke": self.local_smoke,
+                    "platform": self.docker("image", "inspect", "-f", "{{.Os}}/{{.Architecture}}", self.image).strip()})
+        self.docker("network", "create", self.network)
+        self.network_created = True
+        self.round("smoke", 50)
+        if not self.local_smoke and self.per_group != 50:
+            self.round("full", self.per_group)
+
+    def save_container(self, name, label):
+        state = subprocess.run(["docker", "inspect", "-f", "{{json .State}}", name],
+                               capture_output=True, text=True)
+        (self.round_dir / (label + "-state.json")).write_text(state.stdout)
+        result = subprocess.run(["docker", "logs", name], capture_output=True, text=True)
+        (self.round_dir / (label + ".log")).write_text(result.stdout + result.stderr)
+
+    def cleanup(self):
+        for name in self.containers:
+            if self.round_dir:
+                self.save_container(name, name + "-failure")
+            subprocess.run(["docker", "rm", "-f", name], stdout=subprocess.DEVNULL, check=False)
+        if self.network_created:
+            subprocess.run(["docker", "network", "rm", self.network], stdout=subprocess.DEVNULL, check=False)
+        self.summary()
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("command", choices=["run"])
+    parser.add_argument("--local-smoke", action="store_true")
+    args = parser.parse_args()
+    bench = Benchmark(args.local_smoke)
+
+    def cancelled(signum, frame):
+        raise RuntimeError(f"Received signal {signum}")
+
+    signal.signal(signal.SIGTERM, cancelled)
+    try:
+        bench.run()
+    finally:
+        bench.cleanup()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/.github/scripts/metric-snapshot-s3-ops.py b/.github/scripts/metric-snapshot-s3-ops.py
new file mode 100644
index 00000000000..f1e34defb89
--- /dev/null
+++ b/.github/scripts/metric-snapshot-s3-ops.py
@@ -0,0 +1,217 @@
+#!/usr/bin/env python3
+# 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.
+
+# /// script
+# requires-python = ">=3.10"
+# dependencies = ["boto3>=1.35,<2"]
+# ///
+
+"""Budget preflight and strictly scoped S3 operations for the metric snapshot benchmark."""
+
+import argparse
+from decimal import Decimal
+import json
+import os
+from pathlib import Path
+import re
+import time
+import urllib.request
+
+REGION = "ap-southeast-1"
+HOURLY_USD = Decimal("0.9408")
+OTHER_RESERVE_USD = Decimal("1.50")
+
+
+def owned_prefix(prefix):
+    if not re.fullmatch(r"metric-snapshot-bench/[0-9]+-[0-9]+/", prefix):
+        raise ValueError("prefix must be exactly metric-snapshot-bench/<run_id>-<attempt>/")
+    return prefix
+
+
+def plan(env):
+    image = env["GREPTIMEDB_IMAGE"]
+    if not re.fullmatch(r"greptime/greptimedb(?:-nightly)?@sha256:[0-9a-f]{64}", image):
+        raise ValueError("image must be an immutable greptime/greptimedb sha256 reference")
+    if not re.fullmatch(r"[0-9a-f]{40}", env["EXPECTED_COMMIT"]):
+        raise ValueError("expected commit must be a full lowercase source SHA")
+    tables, minutes = int(env["LOGICAL_TABLES_PER_GROUP"]), int(env["BENCHMARK_TIMEOUT_MINUTES"])
+    if not 50 <= tables <= 10_000 or not 30 <= minutes <= 240:
+        raise ValueError("logical tables per group must be 50..10000; timeout must be 30..240 minutes")
+    if env["S3_REGION"] != REGION or env["RUNNER_REGION"] != REGION:
+        raise ValueError("bucket and runner must both use ap-southeast-1")
+    if not env["S3_BUCKET"]:
+        raise ValueError("S3 bucket is required")
+    prior = Decimal(env["PRIOR_SPEND_USD"])
+    # Reserve 45m provision, 10m queue allowance and 20m cleanup beyond the job timeout.
+    reservation = HOURLY_USD * Decimal(minutes + 75) / 60 + OTHER_RESERVE_USD
+    if not prior.is_finite() or prior < 0 or prior >= 15 or prior + reservation > 20:
+        raise ValueError("campaign budget exhausted: prior spend must be below $15 and leave the full attempt reserve within $20")
+    return {"image": image, "expected_commit": env["EXPECTED_COMMIT"],
+            "logical_tables_per_group": tables, "smoke_logical_tables": 500,
+            "timeout_minutes": minutes, "prefix": owned_prefix(env["RUN_PREFIX"]),
+            "instance_type": "c6id.4xlarge", "region": REGION,
+            "prior_estimated_usd": str(prior), "attempt_reserve_usd": str(reservation),
+            "ec2_hourly_usd": str(HOURLY_USD), "other_reserve_usd": str(OTHER_RESERVE_USD),
+            "cost_note": "Estimate, not an AWS billing total or a hard spending cap."}
+
+
+def client(region):
+    import boto3
+    from botocore.config import Config
+
+    if region != REGION:
+        raise ValueError("S3 region must be ap-southeast-1")
+    return boto3.client("s3", region_name=region, config=Config(
+        connect_timeout=10, read_timeout=30, retries={"mode": "standard", "max_attempts": 3}))
+
+
+def entries(s3, operation, bucket, prefix, fields, scope=""):
+    listed = owned_prefix(prefix) + scope
+    for page in s3.get_paginator(operation).paginate(Bucket=bucket, Prefix=listed):
+        for field in fields:
+            for item in page.get(field, []):
+                if not item["Key"].startswith(listed):
+                    raise ValueError("S3 returned an object outside the owned prefix")
+                yield item
+
+
+def inventory(s3, bucket, prefix, round=None):
+    # A round's snapshot lives under <run prefix><round>/; earlier rounds stay until cleanup.
+    if round not in (None, "smoke", "full"):
+        raise ValueError("round must be smoke or full")
+    scope = round + "/" if round else ""
+    objects = list(entries(s3, "list_objects_v2", bucket, prefix, ("Contents",), scope))
+    return {"prefix": prefix + scope, "objects": len(objects), "bytes": sum(item["Size"] for item in objects),
+            "measurement": "S3 listing inventory; not request counts or transferred bytes."}
+
+
+def probe(s3, bucket, prefix):
+    if inventory(s3, bucket, prefix)["objects"]:
+        raise ValueError("run prefix is already occupied; use a fresh workflow attempt")
+    key = prefix + "_probe"
+    s3.put_object(Bucket=bucket, Key=key, Body=b"metric snapshot benchmark")
+    try:
+        body = s3.get_object(Bucket=bucket, Key=key)["Body"]
+        try:
+            if body.read() != b"metric snapshot benchmark":
+                raise ValueError("S3 probe returned unexpected content")
+        finally:
+            body.close()
+    finally:
+        s3.delete_object(Bucket=bucket, Key=key)
+    return inventory(s3, bucket, prefix)
+
+
+def cleanup(s3, bucket, prefix):
+    owned_prefix(prefix)
+    for item in entries(s3, "list_objects_v2", bucket, prefix, ("Contents",)):
+        s3.delete_object(Bucket=bucket, Key=item["Key"])
+    after = inventory(s3, bucket, prefix)
+    if after["objects"]:
+        raise RuntimeError("owned S3 prefix is not empty after cleanup")
+    return after
+
+
+def scrub(root, env):
+    secrets = [env[name].encode() for name in ("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_SESSION_TOKEN")
+               if env.get(name)]
+    if not secrets:
+        raise ValueError("credential scan requires the CI credential environment")
+    overlap = max(map(len, secrets)) - 1
+    for result in Path(root).rglob("*"):
+        if result.is_file():
+            with result.open("rb") as stream:
+                previous = b""
+                while chunk := stream.read(1024 * 1024):
+                    data = previous + chunk
+                    if any(secret in data for secret in secrets):
+                        raise ValueError("credential detected in results; artifact upload is blocked")
+                    previous = data[-overlap:] if overlap else b""
+
+
+def wait_for_benchmark(env, started_at):
+    """Bound runner allocation even when the benchmark job never leaves its queue."""
+    run_id, attempt = owned_prefix(env["RUN_PREFIX"]).split("/")[1].split("-")
+    repo = env["GITHUB_REPOSITORY"]
+    if run_id != env["GITHUB_RUN_ID"] or not re.fullmatch(r"[\w.-]+/[\w.-]+", repo):
+        raise ValueError("workflow identity does not match its owned resource prefix")
+    minutes = int(env["BENCHMARK_TIMEOUT_MINUTES"])
+    if not 30 <= minutes <= 240 or started_at <= 0:
+        raise ValueError("invalid benchmark allocation deadline")
+    # Use the original allocation timestamp; a delayed or retried watcher cannot extend it.
+    deadline = started_at + (45 + minutes + 10) * 60
+    request = urllib.request.Request(
+        f"https://api.github.com/repos/{repo}/actions/runs/{run_id}/attempts/{attempt}/jobs?per_page=100",
+        headers={"Authorization": f"Bearer {env['GH_TOKEN']}", "Accept": "application/vnd.github+json",
+                 "X-GitHub-Api-Version": "2022-11-28"})
+    while True:
+        remaining = deadline - time.time()
+        if remaining <= 0:
+            raise TimeoutError("benchmark allocation deadline reached; proceeding to resource cleanup")
+        # API errors deliberately fail this step: subsequent always() steps clean up immediately.
+        with urllib.request.urlopen(request, timeout=min(15, remaining)) as response:
+            jobs = json.load(response)["jobs"]
+        benchmark = next((job for job in jobs if job["name"] == "benchmark"), None)
+        if benchmark and benchmark["status"] == "completed":
+            return {"benchmark_conclusion": benchmark["conclusion"], "owned_attempt": attempt}
+        remaining = deadline - time.time()
+        if remaining > 0:
+            time.sleep(min(30, remaining))
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("command", choices=("plan", "probe", "inventory", "cleanup", "scrub", "cost", "wait-for-benchmark"))
+    parser.add_argument("--output")
+    parser.add_argument("--started-at", type=int)
+    parser.add_argument("--round")
+    args = parser.parse_args()
+    if args.round is not None and args.command != "inventory":
+        parser.error("--round applies only to inventory")
+    if args.command == "plan":
+        result = plan(os.environ)
+    elif args.command == "scrub":
+        scrub(os.environ["RESULTS"], os.environ)
+        return
+    elif args.command == "wait-for-benchmark":
+        if args.started_at is None:
+            parser.error("wait-for-benchmark requires --started-at")
+        result = wait_for_benchmark(os.environ, args.started_at)
+    elif args.command == "cost":
+        if args.started_at is None:
+            parser.error("cost requires --started-at")
+        seconds = max(0, int(time.time()) - args.started_at)
+        estimate = HOURLY_USD * Decimal(seconds) / 3600
+        result = {"allocation_window_seconds": seconds, "ec2_estimate_usd": str(round(estimate, 4)),
+                  "campaign_estimate_with_other_reserve_usd": str(round(
+                      Decimal(os.environ["PRIOR_SPEND_USD"]) + estimate + OTHER_RESERVE_USD, 4)),
+                  "note": "Allocation wall-time estimate plus $1.50 reserve for S3, disk and IPv4; not an AWS bill."}
+    else:
+        prefix = owned_prefix(os.environ["RUN_PREFIX"])
+        s3, bucket = client(os.environ["S3_REGION"]), os.environ["S3_BUCKET"]
+        if args.command == "inventory":
+            result = inventory(s3, bucket, prefix, args.round)
+        else:
+            result = globals()[args.command](s3, bucket, prefix)
+    rendered = json.dumps(result, indent=2) + "\n"
+    if args.output:
+        destination = Path(args.output)
+        destination.parent.mkdir(parents=True, exist_ok=True)
+        destination.write_text(rendered)
+    print(rendered, end="")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/.github/scripts/tests/test_metric_snapshot_fixture.py b/.github/scripts/tests/test_metric_snapshot_fixture.py
new file mode 100644
index 00000000000..f4c16a11af3
--- /dev/null
+++ b/.github/scripts/tests/test_metric_snapshot_fixture.py
@@ -0,0 +1,101 @@
+# 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.
+
+import copy
+import importlib.util
+from pathlib import Path
+import unittest
+
+spec = importlib.util.spec_from_file_location(
+    "fixture", Path(__file__).resolve().parents[1] / "metric-snapshot-fixture.py"
+)
+fixture = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(fixture)
+
+
+class ValidatorTests(unittest.TestCase):
+    def records(self):
+        return {
+            "schema": {"column_schemas": [
+                {"name": "ts", "data_type": "TimestampMillisecond"},
+                *[{"name": f"tag_{i:03d}", "data_type": "String"} for i in range(10)],
+                {"name": "val", "data_type": "Float64"},
+            ]},
+            "rows": [fixture.row(0, j, 50) for j in range(32)],
+        }
+
+    def test_full_tag_coverage_and_heterogeneous_schemas(self):
+        for per_group in (50, 10000):
+            variants = {tuple(fixture.tags(i, per_group)) for i in range(per_group)}
+            self.assertEqual(set().union(*map(set, variants)), set(range(298)))
+            self.assertGreater(len(variants), 1)
+            self.assertTrue(all(len(tags) == 10 for tags in variants))
+
+    def test_accepts_engine_column_order_with_matching_values(self):
+        records = self.records()
+        order = [*range(1, 11), 0, 11]  # The engine returns tags, timestamp, value.
+        records["schema"]["column_schemas"] = [
+            records["schema"]["column_schemas"][index] for index in order
+        ]
+        records["rows"] = [[values[index] for index in order] for values in records["rows"]]
+        fixture.check_records(0, 50, records)
+        records["rows"][1][10] = 2
+        with self.assertRaisesRegex(RuntimeError, "typed values"):
+            fixture.check_records(0, 50, records)
+
+    def test_rejects_column_type_and_membership_changes(self):
+        fixture.check_records(0, 50, self.records())
+        for mutation in ("type", "extra", "missing"):
+            records = self.records()
+            columns = records["schema"]["column_schemas"]
+            if mutation == "type":
+                columns[0]["data_type"] = "TimestampMicrosecond"
+            elif mutation == "extra":
+                columns.append({"name": "unexpected", "data_type": "String"})
+            else:
+                del columns[1]
+            with self.subTest(mutation=mutation), self.assertRaisesRegex(RuntimeError, "schema"):
+                fixture.check_records(0, 50, records)
+
+    def test_rejects_null_and_numeric_type_changes(self):
+        for value in ("", 0, False):
+            records = self.records()
+            records["rows"][0][-1] = value  # Expected SQL NULL.
+            with self.subTest(value=value), self.assertRaisesRegex(RuntimeError, "typed values"):
+                fixture.check_records(0, 50, records)
+        records = self.records()
+        records["rows"][4][-1] = 1  # Numeric equality to 1.0 must not hide a type change.
+        with self.assertRaisesRegex(RuntimeError, "typed values"):
+            fixture.check_records(0, 50, records)
+
+    def test_source_comparison_requires_same_ddl_and_remapped_ids(self):
+        source = {"fixture": {"tables_per_group": 50}, "schema_sha256": "schema",
+                  "data_sha256": "data", "ddl_sha256": "ddl",
+                  "table_ids": {"m000000": 1024, "phy_00": 1025}}
+        target = copy.deepcopy(source)
+        target["table_ids"] = {"m000000": 1041, "phy_00": 1042}
+        fixture.compare_source(target, source)
+        self.assertEqual(target["remapped_table_ids"], 2)
+        for field in ("schema_sha256", "data_sha256", "ddl_sha256"):
+            changed = copy.deepcopy(target)
+            changed[field] = "changed"
+            with self.subTest(field=field), self.assertRaisesRegex(RuntimeError, field):
+                fixture.compare_source(changed, source)
+        target["table_ids"]["phy_00"] = 1025
+        with self.assertRaisesRegex(RuntimeError, "not all remapped"):
+            fixture.compare_source(target, source)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/.github/scripts/tests/test_metric_snapshot_s3_bench.py b/.github/scripts/tests/test_metric_snapshot_s3_bench.py
new file mode 100644
index 00000000000..c662ecbff76
--- /dev/null
+++ b/.github/scripts/tests/test_metric_snapshot_s3_bench.py
@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Failure accounting for the metric snapshot runner."""
+import importlib.util
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+SPEC = importlib.util.spec_from_file_location(
+    "metric_snapshot_bench", Path(__file__).resolve().parents[1] / "metric-snapshot-s3-bench.py")
+bench = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(bench)
+
+
+class PhaseTests(unittest.TestCase):
+    def test_docker_nanosecond_timestamps(self):
+        start = bench.docker_timestamp("2026-10-14T06:06:44.72000818Z")
+        end = bench.docker_timestamp("2026-10-14T06:06:46.123456789Z")
+        self.assertAlmostEqual((end - start).total_seconds(), 1.403448)
+
+    def test_resource_interference_invalidates_success_and_preserves_fixture(self):
+        with tempfile.TemporaryDirectory() as directory:
+            runner = bench.Benchmark.__new__(bench.Benchmark)
+            runner.round_dir = Path(directory)
+            runner.server_pid = 0
+            runner.phases = []
+            runner.metrics = lambda: ""
+            runner.summary = lambda: None
+            fixture = runner.round_dir / "source-validation.json"
+            fixture.write_text('{"data_sha256":"fixture"}')
+            before = {"available": True, "pswpout_pages": 0, "oom_kill": 0}
+            after = {"available": True, "pswpout_pages": 1, "oom_kill": 0}
+            with patch.object(bench, "host_memory", side_effect=[before, after]):
+                with self.assertRaisesRegex(RuntimeError, "Resource interference"):
+                    runner.phase("source-validation", lambda: None)
+            self.assertEqual(json.loads(fixture.read_text()), {"data_sha256": "fixture"})
+            evidence = json.loads((runner.round_dir / "source-validation-resources.json").read_text())
+            self.assertEqual(evidence["outcome"], "resource-invalid")
+            self.assertEqual(evidence["new_swapout_pages"], 1)
+
+    def test_action_failure_keeps_failed_resource_evidence(self):
+        with tempfile.TemporaryDirectory() as directory:
+            runner = bench.Benchmark.__new__(bench.Benchmark)
+            runner.round_dir = Path(directory)
+            runner.server_pid = 0
+            runner.phases = []
+            runner.metrics = lambda: ""
+            runner.summary = lambda: None
+            def fail():
+                raise RuntimeError("export failed")
+            with patch.object(bench, "host_memory", return_value={"available": False}):
+                with self.assertRaisesRegex(RuntimeError, "export failed"):
+                    runner.phase("export", fail)
+            evidence = json.loads((runner.round_dir / "export-resources.json").read_text())
+            self.assertEqual(evidence["outcome"], "failed")
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/.github/scripts/tests/test_metric_snapshot_s3_ops.py b/.github/scripts/tests/test_metric_snapshot_s3_ops.py
new file mode 100644
index 00000000000..d22c09044b2
--- /dev/null
+++ b/.github/scripts/tests/test_metric_snapshot_s3_ops.py
@@ -0,0 +1,203 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Offline safety checks for the metric snapshot workflow's budget and cleanup."""
+
+import importlib.util
+import io
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest import mock
+
+SPEC = importlib.util.spec_from_file_location("snapshot_ops", Path(__file__).parents[1] / "metric-snapshot-s3-ops.py")
+ops = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(ops)
+PREFIX = "metric-snapshot-bench/123-1/"
+
+
+class FakeS3:
+    def __init__(self):
+        self.objects = [{"Key": PREFIX + "current", "Size": 11}, {"Key": PREFIX + "data", "Size": 7}]
+        self.deleted = []
+
+    def get_paginator(self, operation):
+        assert operation == "list_objects_v2"
+        parent = self
+
+        class Paginator:
+            def paginate(self, **kwargs):
+                assert kwargs["Bucket"] == "bucket" and kwargs["Prefix"].startswith(PREFIX)
+                # Keys outside the run prefix model a misbehaving listing and are always returned.
+                return iter([{"Contents": [item for item in parent.objects if not item["Key"].startswith(PREFIX)
+                                           or item["Key"].startswith(kwargs["Prefix"])]}])
+
+        return Paginator()
+
+    def put_object(self, **kwargs):
+        self.objects.append({"Key": kwargs["Key"], "Size": len(kwargs["Body"])})
+        return {}
+
+    def get_object(self, **kwargs):
+        return {"Body": io.BytesIO(b"metric snapshot benchmark")}
+
+    def delete_object(self, **kwargs):
+        self.deleted.append(kwargs)
+        self.objects = [item for item in self.objects if item["Key"] != kwargs["Key"]]
+
+
+class OperationsTest(unittest.TestCase):
+    def env(self, **changes):
+        return dict(GREPTIMEDB_IMAGE="greptime/greptimedb-nightly@sha256:" + "a" * 64,
+                    EXPECTED_COMMIT="b" * 40, LOGICAL_TABLES_PER_GROUP="10000",
+                    BENCHMARK_TIMEOUT_MINUTES="240", PRIOR_SPEND_USD="0",
+                    S3_REGION="ap-southeast-1", RUNNER_REGION="ap-southeast-1", S3_BUCKET="bucket",
+                    RUN_PREFIX=PREFIX) | changes
+
+    def test_budget_includes_provision_cleanup_and_other_cost_reserve(self):
+        plan = ops.plan(self.env())
+        self.assertEqual(plan["attempt_reserve_usd"], "6.4392")
+        ops.plan(self.env(PRIOR_SPEND_USD="13.5608"))
+        for changes in ({"PRIOR_SPEND_USD": "13.5609"}, {"PRIOR_SPEND_USD": "15"},
+                        {"PRIOR_SPEND_USD": "NaN"}, {"PRIOR_SPEND_USD": "-1"},
+                        {"BENCHMARK_TIMEOUT_MINUTES": "241"}, {"LOGICAL_TABLES_PER_GROUP": "49"},
+                        {"LOGICAL_TABLES_PER_GROUP": "10001"}, {"S3_REGION": "us-east-1"},
+                        {"GREPTIMEDB_IMAGE": "greptime/greptimedb:latest"},
+                        {"GREPTIMEDB_IMAGE": "docker.io/greptime/greptimedb@sha256:" + "a" * 64},
+                        {"GREPTIMEDB_IMAGE": "greptime/greptimedb:nightly@sha256:" + "a" * 64},
+                        {"EXPECTED_COMMIT": "abcdef"}):
+            with self.subTest(changes=changes), self.assertRaises(ValueError):
+                ops.plan(self.env(**changes))
+
+    def watcher_env(self):
+        return self.env(GITHUB_REPOSITORY="GreptimeTeam/greptimedb", GITHUB_RUN_ID="123",
+                        GITHUB_RUN_ATTEMPT="2", GH_TOKEN="test-token", BENCHMARK_TIMEOUT_MINUTES="30")
+
+    def test_watcher_completed_job_uses_original_attempt_and_does_not_wait(self):
+        for conclusion in ("success", "failure", "cancelled"):
+            response = io.BytesIO(json.dumps({"jobs": [{"name": "benchmark", "status": "completed",
+                                                       "conclusion": conclusion}]}).encode())
+            with self.subTest(conclusion=conclusion), \
+                    mock.patch.object(ops.urllib.request, "urlopen", return_value=response) as fetch, \
+                    mock.patch.object(ops.time, "time", return_value=1005), \
+                    mock.patch.object(ops.time, "sleep") as sleep:
+                self.assertEqual(ops.wait_for_benchmark(self.watcher_env(), 1000),
+                                 {"benchmark_conclusion": conclusion, "owned_attempt": "1"})
+                self.assertIn("/runs/123/attempts/1/jobs?", fetch.call_args.args[0].full_url)
+                sleep.assert_not_called()
+
+    def test_watcher_queue_reaches_durable_deadline(self):
+        deadline = 1000 + (45 + 30 + 10) * 60
+        clock = [deadline - 31]
+
+        def advance(seconds):
+            self.assertLessEqual(seconds, 30)
+            clock[0] += seconds
+
+        def queued(*args, **kwargs):
+            return io.BytesIO(b'{"jobs": [{"name": "benchmark", "status": "queued"}]}')
+
+        with mock.patch.object(ops.urllib.request, "urlopen", side_effect=queued) as fetch, \
+                mock.patch.object(ops.time, "time", side_effect=lambda: clock[0]), \
+                mock.patch.object(ops.time, "sleep", side_effect=advance), \
+                self.assertRaisesRegex(TimeoutError, "proceeding to resource cleanup"):
+            ops.wait_for_benchmark(self.watcher_env(), 1000)
+        self.assertEqual(clock[0], deadline)
+        self.assertEqual(fetch.call_count, 2)
+
+    def test_watcher_api_error_fails_immediately_for_cleanup(self):
+        with mock.patch.object(ops.urllib.request, "urlopen", side_effect=OSError("API unavailable")), \
+                mock.patch.object(ops.time, "time", return_value=1005), \
+                mock.patch.object(ops.time, "sleep") as sleep, \
+                self.assertRaises(OSError):
+            ops.wait_for_benchmark(self.watcher_env(), 1000)
+        sleep.assert_not_called()
+
+    def test_cleanup_deletes_every_listed_object(self):
+        s3 = FakeS3()
+        result = ops.cleanup(s3, "bucket", PREFIX)
+        self.assertEqual(result["objects"], 0)
+        self.assertEqual(s3.deleted, [{"Bucket": "bucket", "Key": PREFIX + "current"},
+                                      {"Bucket": "bucket", "Key": PREFIX + "data"}])
+
+    def test_probe_refuses_occupied_prefix_and_removes_its_object(self):
+        with self.assertRaisesRegex(ValueError, "already occupied"):
+            ops.probe(FakeS3(), "bucket", PREFIX)
+        s3 = FakeS3()
+        s3.objects = []
+        self.assertEqual(ops.probe(s3, "bucket", PREFIX)["objects"], 0)
+        self.assertEqual(s3.deleted, [{"Bucket": "bucket", "Key": PREFIX + "_probe"}])
+
+    def test_failed_probe_delete_still_has_hosted_cleanup_path(self):
+        s3 = FakeS3()
+        s3.objects = []
+        with mock.patch.object(s3, "delete_object", side_effect=OSError("temporary delete failure")), \
+                self.assertRaises(OSError):
+            ops.probe(s3, "bucket", PREFIX)
+        self.assertEqual([item["Key"] for item in s3.objects], [PREFIX + "_probe"])
+        workflow = (Path(__file__).parents[2] / "workflows/metric-snapshot-s3-bench.yml").read_text()
+        teardown = workflow.split("  teardown:\n", 1)[1]
+        job_if = next(line.strip() for line in teardown.splitlines() if line.startswith("    if:"))
+        self.assertEqual(job_if, "if: ${{ always() && needs.validate.outputs.run_id != '' }}")
+        ec2 = teardown.split("    - name: Delete EC2, its temporary disks and runner registration\n", 1)[1]
+        self.assertEqual(ec2.splitlines()[0].strip(), "if: ${{ always() && needs.provision.result != 'skipped' }}")
+        self.assertIn("      RUN_PREFIX: ${{ needs.validate.outputs.run_prefix }}\n", teardown.split("    steps:", 1)[0])
+        s3_cleanup = teardown.split("    - name: Remove and verify only this run's S3 prefix\n", 1)[1].split("    - ", 1)[0]
+        self.assertEqual(s3_cleanup.splitlines()[0].strip(), "if: always()")
+        self.assertIn("run: uv run greptimedb/.github/scripts/metric-snapshot-s3-ops.py cleanup\n", s3_cleanup)
+        self.assertEqual(ops.cleanup(s3, "bucket", PREFIX)["objects"], 0)
+
+    def test_round_inventory_excludes_other_rounds_while_cleanup_covers_the_run(self):
+        s3 = FakeS3()
+        s3.objects = [{"Key": PREFIX + "smoke/pack-000000.bin", "Size": 3},
+                      {"Key": PREFIX + "full/pack-000000.bin", "Size": 5},
+                      {"Key": PREFIX + "full/index.json", "Size": 2}]
+        full = ops.inventory(s3, "bucket", PREFIX, "full")
+        self.assertEqual((full["prefix"], full["objects"], full["bytes"]), (PREFIX + "full/", 2, 7))
+        self.assertEqual(ops.inventory(s3, "bucket", PREFIX, "smoke")["bytes"], 3)
+        self.assertEqual(ops.inventory(s3, "bucket", PREFIX)["objects"], 3)
+        for round in ("", "../full", "full/"):
+            with self.subTest(round=round), self.assertRaises(ValueError):
+                ops.inventory(s3, "bucket", PREFIX, round)
+        self.assertEqual(ops.cleanup(s3, "bucket", PREFIX)["objects"], 0)
+        self.assertEqual(len(s3.deleted), 3)
+
+    def test_bad_prefix_and_out_of_scope_response_cannot_delete(self):
+        for prefix in ("", "metric-snapshot-bench/", "metric-snapshot-bench/123-1", PREFIX + "../", "oswal-bench/123-1/"):
+            s3 = FakeS3()
+            with self.subTest(prefix=prefix), self.assertRaises(ValueError):
+                ops.cleanup(s3, "bucket", prefix)
+            self.assertEqual(s3.deleted, [])
+        s3 = FakeS3()
+        s3.objects[0]["Key"] = "other-run/current"
+        with self.assertRaises(ValueError):
+            ops.inventory(s3, "bucket", PREFIX, "full")
+        with self.assertRaises(ValueError):
+            ops.cleanup(s3, "bucket", PREFIX)
+        self.assertEqual(s3.deleted, [])
+
+    def test_credential_crossing_read_boundary_blocks_artifact_upload(self):
+        with tempfile.TemporaryDirectory() as directory:
+            logfile = Path(directory) / "server.log"
+            logfile.write_bytes(b"a" * (1024 * 1024 - 4) + b"private-key" + b"tail")
+            with self.assertRaisesRegex(ValueError, "upload is blocked"):
+                ops.scrub(directory, {"AWS_SECRET_ACCESS_KEY": "private-key"})
+            logfile.write_text("safe log\n")
+            ops.scrub(directory, {"AWS_SECRET_ACCESS_KEY": "private-key"})
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml
index deba5010c80..2aa460ec0ca 100644
--- a/.github/workflows/checks.yml
+++ b/.github/workflows/checks.yml
@@ -88,6 +88,8 @@ jobs:
         run: .github/scripts/run-fuzz-targets-test.sh
       - name: Run CI comment-command tests
         run: python3 .github/scripts/tests/test_ci_slash.py
+      - name: Run metric snapshot benchmark tests
+        run: python3 -m unittest discover -s .github/scripts/tests -p 'test_metric_snapshot*.py'
       - name: Run query-regression Python tooling tests
         run: |
           python3 tests/perf/test_query_regression_runner_compaction_toctou.py
diff --git a/.github/workflows/metric-snapshot-s3-bench.yml b/.github/workflows/metric-snapshot-s3-bench.yml
new file mode 100644
index 00000000000..e4a12dde69f
--- /dev/null
+++ b/.github/workflows/metric-snapshot-s3-bench.yml
@@ -0,0 +1,230 @@
+name: Metric Snapshot S3 Benchmark
+on:
+  workflow_dispatch:
+    inputs:
+      greptimedb_image:
+        description: 'Immutable greptime/greptimedb or greptimedb-nightly image (repository@sha256:digest)'
+        required: true
+        type: string
+      expected_commit:
+        description: 'Full source commit SHA reported by the image'
+        required: true
+        type: string
+      logical_tables_per_group:
+        description: 'Logical tables per physical table, 50..10000; 10 physical tables, 32 rows per logical table'
+        type: number
+        default: 10000
+      benchmark_timeout_minutes:
+        description: 'Setup, 500-table smoke and full run timeout, 30..240 minutes'
+        type: number
+        default: 240
+      prior_spend_usd:
+        description: 'Cumulative estimated cloud spend for all earlier attempts in this $20 campaign'
+        required: true
+        type: string
+        default: '0'
+permissions:
+  contents: read
+concurrency:
+  group: metric-snapshot-s3-bench
+  cancel-in-progress: false
+env:
+  GREPTIMEDB_IMAGE: ${{ inputs.greptimedb_image }}
+  EXPECTED_COMMIT: ${{ inputs.expected_commit }}
+  LOGICAL_TABLES_PER_GROUP: ${{ inputs.logical_tables_per_group }}
+  BENCHMARK_TIMEOUT_MINUTES: ${{ inputs.benchmark_timeout_minutes }}
+  PRIOR_SPEND_USD: ${{ inputs.prior_spend_usd }}
+  RUN_PREFIX: metric-snapshot-bench/${{ github.run_id }}-${{ github.run_attempt }}/
+  S3_BUCKET: ${{ vars.AWS_CI_TEST_BUCKET }}
+  S3_REGION: ${{ vars.AWS_CI_TEST_BUCKET_REGION }}
+  RUNNER_REGION: ${{ vars.EC2_RUNNER_REGION }}
+jobs:
+  validate:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    outputs:
+      instance_type: ${{ steps.runner.outputs.instance_type }}
+      run_id: ${{ steps.identity.outputs.run_id }}
+      run_prefix: ${{ steps.identity.outputs.run_prefix }}
+    steps:
+    - uses: actions/checkout@v4
+      with:
+        path: greptimedb
+        persist-credentials: false
+    - name: Test benchmark boundaries
+      run: |
+        python3 greptimedb/.github/scripts/tests/test_metric_snapshot_s3_ops.py
+        python3 greptimedb/.github/scripts/tests/test_metric_snapshot_fixture.py
+        python3 greptimedb/.github/scripts/tests/test_metric_snapshot_s3_bench.py
+    - name: Validate pinned version, scale, region and cumulative budget
+      run: python3 greptimedb/.github/scripts/metric-snapshot-s3-ops.py plan
+    - name: Bind resource ownership to this validated attempt
+      id: identity
+      run: |
+        echo "run_id=$GITHUB_RUN_ID-$GITHUB_RUN_ATTEMPT" >> "$GITHUB_OUTPUT"
+        echo "run_prefix=$RUN_PREFIX" >> "$GITHUB_OUTPUT"
+    - uses: astral-sh/setup-uv@v6
+    - name: Check S3 access before allocating EC2
+      env:
+        AWS_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }}
+        AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }}
+      run: uv run greptimedb/.github/scripts/metric-snapshot-s3-ops.py probe
+    - name: Resolve the fixed benchmark runner
+      id: runner
+      run: python3 greptimedb/.github/scripts/ci-runner-config.py --provider AWS --instance-type c6id.4xlarge
+  provision:
+    needs: validate
+    runs-on: ubuntu-latest
+    timeout-minutes: 45
+    outputs:
+      label: ${{ steps.aws.outputs.label }}
+      instance_id: ${{ steps.aws.outputs.instance_id }}
+      runner_name: ${{ steps.aws.outputs.runner_name }}
+      started_at: ${{ steps.start.outputs.started_at }}
+    steps:
+    - uses: actions/checkout@v4
+      with:
+        path: greptimedb
+        persist-credentials: false
+    - uses: aws-actions/configure-aws-credentials@v4
+      with:
+        aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
+        aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
+        aws-region: ${{ env.RUNNER_REGION }}
+    - uses: astral-sh/setup-uv@v6
+    - name: Record allocation start
+      id: start
+      run: echo "started_at=$(date +%s)" >> "$GITHUB_OUTPUT"
+    - name: Create ephemeral EC2 runner
+      id: aws
+      env:
+        GH_PERSONAL_ACCESS_TOKEN: ${{ secrets.GH_PERSONAL_ACCESS_TOKEN }}
+        AWS_REGION: ${{ env.RUNNER_REGION }}
+        AWS_EC2_IMAGE_ID: ${{ vars.BENCHMARK_EC2_IMAGE_ID || 'auto' }}
+        AWS_EC2_INSTANCE_TYPE: ${{ needs.validate.outputs.instance_type }}
+        AWS_EC2_SUBNET_ID: ${{ vars.EC2_RUNNER_SUBNET_ID }}
+        AWS_EC2_SECURITY_GROUP_ID: ${{ vars.EC2_RUNNER_SECURITY_GROUP_ID }}
+        AWS_EC2_SYSTEM_DISK_GIB: '60'
+        AWS_EC2_TTL_HOURS: '6'
+        AWS_EC2_RUN_ID: ${{ needs.validate.outputs.run_id }}
+      run: uv run greptimedb/.github/scripts/aws-ec2-runner-provision.py
+  benchmark:
+    needs:
+    - validate
+    - provision
+    runs-on: ${{ needs.provision.outputs.label }}
+    timeout-minutes: ${{ fromJSON(format('{0}', inputs.benchmark_timeout_minutes)) }}
+    env:
+      RUN_PREFIX: ${{ needs.validate.outputs.run_prefix }}
+      RESULTS: ${{ github.workspace }}/metric-snapshot-results
+      DISK_ROOT: /mnt/metric-snapshot-disk
+      DATA_ROOT: /mnt/metric-snapshot-disk/data
+    steps:
+    - uses: actions/checkout@v4
+      with:
+        path: greptimedb
+        persist-credentials: false
+    - name: Set up benchmark dependencies
+      uses: ./greptimedb/.github/actions/setup-benchmark
+    - uses: astral-sh/setup-uv@v6
+    - name: Mount the temporary local File data disk
+      run: |
+        device=$(lsblk -dnpo NAME,MODEL | awk '/Instance Storage/ {print $1; exit}')
+        test -n "$device"
+        if [[ "$(lsblk -nro TYPE "$device" | wc -l)" -ne 1 ]] || findmnt --source "$device" >/dev/null; then
+          echo 'Refusing partitioned or mounted instance storage' >&2
+          exit 1
+        fi
+        probe_status=0
+        sudo -n blkid -p "$device" || probe_status=$?
+        if ((probe_status != 2)); then
+          echo 'Refusing a disk with an existing signature or an unsuccessful signature probe' >&2
+          exit 1
+        fi
+        sudo -n mkfs.ext4 -q "$device"
+        sudo -n mkdir -p "$DISK_ROOT"
+        sudo -n mount -o noatime "$device" "$DISK_ROOT"
+        sudo -n chown "$(id -u):$(id -g)" "$DISK_ROOT"
+        findmnt -no SOURCE,FSTYPE -T "$DISK_ROOT"
+    - name: Probe the real same-region S3 bucket
+      env:
+        AWS_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }}
+        AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }}
+      run: uv run greptimedb/.github/scripts/metric-snapshot-s3-ops.py probe
+    - name: Run the 500-table smoke, then one full packed export and restore
+      env:
+        AWS_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }}
+        AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }}
+      run: python3 greptimedb/.github/scripts/metric-snapshot-s3-bench.py run
+    - name: Check results for credentials
+      id: scrub
+      if: always()
+      env:
+        AWS_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }}
+        AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }}
+      run: python3 greptimedb/.github/scripts/metric-snapshot-s3-ops.py scrub
+    - name: Publish the readable result
+      if: ${{ always() && steps.scrub.outcome == 'success' }}
+      run: |
+        if test -f "$RESULTS/summary.md"; then
+          cat "$RESULTS/summary.md" >> "$GITHUB_STEP_SUMMARY"
+        fi
+    - name: Upload bounded-retention evidence
+      if: ${{ always() && steps.scrub.outcome == 'success' }}
+      uses: actions/upload-artifact@v4
+      with:
+        name: metric-snapshot-s3-${{ github.run_id }}-${{ github.run_attempt }}
+        path: ${{ env.RESULTS }}
+        if-no-files-found: warn
+        retention-days: 14
+  teardown:
+    needs:
+    - validate
+    - provision
+    # Start alongside the benchmark so an offline runner cannot hold cleanup in its queue.
+    if: ${{ always() && needs.validate.outputs.run_id != '' }}
+    runs-on: ubuntu-latest
+    timeout-minutes: 330
+    permissions:
+      contents: read
+      actions: read
+    env:
+      RUN_PREFIX: ${{ needs.validate.outputs.run_prefix }}
+    steps:
+    - uses: actions/checkout@v4
+      with:
+        path: greptimedb
+        persist-credentials: false
+    - uses: astral-sh/setup-uv@v6
+    - name: Wait for completion within the allocation deadline
+      if: ${{ !cancelled() && needs.provision.result == 'success' && needs.provision.outputs.started_at != '' }}
+      env:
+        GH_TOKEN: ${{ github.token }}
+        STARTED_AT: ${{ needs.provision.outputs.started_at }}
+      run: python3 greptimedb/.github/scripts/metric-snapshot-s3-ops.py wait-for-benchmark --started-at "$STARTED_AT"
+    - uses: aws-actions/configure-aws-credentials@v4
+      if: ${{ always() && needs.provision.result != 'skipped' }}
+      with:
+        aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
+        aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
+        aws-region: ${{ env.RUNNER_REGION }}
+    - name: Delete EC2, its temporary disks and runner registration
+      if: ${{ always() && needs.provision.result != 'skipped' }}
+      env:
+        GH_PERSONAL_ACCESS_TOKEN: ${{ secrets.GH_PERSONAL_ACCESS_TOKEN }}
+        AWS_REGION: ${{ env.RUNNER_REGION }}
+        AWS_EC2_INSTANCE_ID: ${{ needs.provision.outputs.instance_id }}
+        AWS_EC2_RUNNER_NAME: ${{ needs.provision.outputs.runner_name }}
+        AWS_EC2_RUN_ID: ${{ needs.validate.outputs.run_id }}
+      run: uv run greptimedb/.github/scripts/aws-ec2-runner-teardown.py
+    - name: Remove and verify only this run's S3 prefix
+      if: always()
+      env:
+        AWS_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }}
+        AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }}
+      run: uv run greptimedb/.github/scripts/metric-snapshot-s3-ops.py cleanup
+    - name: Estimate the cumulative cloud cost
+      if: ${{ always() && needs.provision.outputs.started_at != '' }}
+      env:
+        STARTED_AT: ${{ needs.provision.outputs.started_at }}
+      run: python3 greptimedb/.github/scripts/metric-snapshot-s3-ops.py cost --started-at "$STARTED_AT" >> "$GITHUB_STEP_SUMMARY"
diff --git a/docs/how-to/metric-snapshot-s3-bench.md b/docs/how-to/metric-snapshot-s3-bench.md
new file mode 100644
index 00000000000..86468bf3ffd
--- /dev/null
+++ b/docs/how-to/metric-snapshot-s3-bench.md
@@ -0,0 +1,79 @@
+# Metric snapshot benchmark on AWS S3
+
+The manual `Metric Snapshot S3 Benchmark` workflow tests packed Metric export and
+restore against real S3 in `ap-southeast-1`. It must first exist on the repository's
+default branch for GitHub to register `workflow_dispatch`. It was validated on AWS
+by run [38544464731](https://github.com/GreptimeTeam/greptimedb/actions/runs/38544464731);
+[#9573](https://github.com/GreptimeTeam/greptimedb/pull/9573) records its conditions and results.
+
+## Run
+
+Use the repository's existing EC2 runner configuration and CI S3 credentials.
+The workflow passes credentials only through CI; no local AWS credentials are
+needed. Source, target and CLI use the same immutable image. This image's source
+SHA and required CLI capabilities have been checked:
+
+```bash
+gh workflow run metric-snapshot-s3-bench.yml \
+  --repo GreptimeTeam/greptimedb --ref main \
+  -f greptimedb_image=greptime/greptimedb-nightly@sha256:20a1182bed0f452e6d573ca110b313e0b5cb76de7aedfa41a0c339963d4683d3 \
+  -f expected_commit=f690bbb2040865c0d22b0b57d51612b60ae27cd0 \
+  -f logical_tables_per_group=10000 \
+  -f benchmark_timeout_minutes=240 \
+  -f prior_spend_usd=0
+```
+
+Start a new dispatch for each attempt and update `prior_spend_usd` to include all
+earlier cloud attempts in the campaign, including failed setup and cleanup.
+Do not reset the estimate to zero when retrying. The workflow accepts 50–10,000
+logical tables per group and a 30–240 minute benchmark job timeout.
+
+The runner is fixed at `c6id.4xlarge` (16 vCPU, 32 GiB RAM). At the recorded
+Singapore Linux on-demand estimate of $0.9408/hour, the default attempt reserves
+$6.4392: four benchmark hours, 45 minutes of provisioning, 10 minutes for queueing,
+20 minutes of cleanup, and $1.50 for S3, disk and IPv4 costs. Preflight rejects prior spend at or above
+$15, or an attempt whose reserve would exceed the cumulative $20 budget.
+These estimates and checks are not an AWS hard spending cap or an AWS bill.
+
+## Workload and validation
+
+- A 10-physical-table × 50-logical-table smoke runs first. Its 500 logical tables
+  cover all 298 string tag columns before the larger workload starts.
+- The default full run uses 10 physical tables and 100,000 logical tables, with
+  32 rows per logical table: 3.2 million rows. Each logical schema has 10
+  overlapping tags, a timestamp and a DOUBLE; schemas vary between tables.
+- Source and target use local File storage on the temporary NVMe disk. Only
+  snapshots use S3. Source validation finishes before the source process stops;
+  restoration then starts a separate target process.
+- Export uses `--pack-metric-data`, parallelism 4, chunk parallelism 1, and one
+  complete time chunk. Import uses task parallelism 1 and automatic batch DDL.
+- Every source and restored logical table is checked for membership, schema,
+  DDL and typed values including NULLs. Target placeholders force table ID
+  remapping; successful batch DDL and removal of import resume state are checked.
+
+## Results and cleanup
+
+The job summary reports complete export, structural verify and import durations;
+data generation and full data validation are timed separately. Evidence artifacts
+are credential-scanned and retained for 14 days. Keep conclusions limited to
+this version's absolute AWS results; this workflow does not compare old versions.
+
+Artifacts include sampled CPU/RSS, host memory, swap/OOM counters, server metrics,
+CLI logs and S3 object/byte inventories. Configured swap is not RAM. A phase with
+new swap-out or OOM fails and retains its evidence. OpenDAL operation counters
+are not complete AWS HTTP request counts: multipart calls, retries and CLI traffic
+are not fully covered. Object inventory bytes are not transferred bytes.
+
+A separate GitHub-hosted teardown job starts when provisioning finishes and polls
+the benchmark job. Its deadline includes queue time: allocation start plus 45
+minutes, the benchmark timeout and 10 minutes. Completion, cancellation, an API
+error or the deadline leads to cleanup. Failed provisioning skips the wait.
+It terminates the EC2 instance and its disks, unregisters the runner, deletes every
+object listed under the run's unique prefix and verifies the listing is empty. Like
+the object store WAL benchmark, it needs only list, put, get and delete on the CI
+bucket. It does not abort incomplete multipart uploads or remove noncurrent object
+versions; those depend on the bucket's lifecycle configuration. Check teardown completion before another
+attempt. A deadline may leave the GitHub benchmark job queued after its resources
+are removed; cancel that workflow manually. If cleanup fails, rerun the teardown
+job only; it keeps the original attempt's resource identity and deadline. A new benchmark needs a new dispatch because teardown
+removes its runner. The daily EC2 janitor is only a fallback and does not clean S3.