Merge upstream/main into narthex (pre-merge rehearsal, not for push)

Rehearsal merge of main (f8d8fe1071) into narthex (115d85b669) on a throwaway
local branch. 31 semantic conflicts resolved; workspace and all test targets
compile; promql 1215/1215 and mito2 series_index 185/185 pass.

Pre-merge pins and follow-ups are marked with TODO(narthex-merge):
- greptime-proto is temporarily pinned to the fork branch carrying
  GreptimeTeam/greptime-proto#355; replace with the upstream rev once it lands.
- native payload byte cache is disabled (main removed PageRangeCache); two
  samples tests are #[ignore]d until it gets a dedicated page-cache budget.
- columnar metadata now has its own CacheManager tier (a quarter of
  sst_meta_cache_size); the unified decoded tier was not kept.
- query.experimental_native_promql is registered as a query option, otherwise
  the native route is silently rejected by the option snapshot.

Signed-off-by: discord9 <discord9@163.com>

.github/scripts/agent-observability-summary.py

@@ -133,9 +133,10 @@ def benchmark(root, targets):
     return '\n'.join(lines)
 
 
-def lifecycle(root, targets):
-    manifest = read(root / 'run-manifest.json')
-    corpus = read(root / 'corpus' / 'summary.json')
+def lifecycle(root, targets, data_root=None):
+    data_root = data_root or root
+    manifest = read(data_root / 'run-manifest.json')
+    corpus = read(data_root / 'corpus' / 'summary.json')
     lines = ['# Load / lifecycle summary', '',
              f'Dataset: **{cell(manifest.get("profile", corpus.get("data_profile", {}).get("id")))}**; rows: **{cell(corpus.get("line_count"))}**; '
              f'seed: `{cell(corpus.get("seed"))}`.',
@@ -161,12 +162,13 @@ def lifecycle(root, targets):
     physical_rows = []
     load_rows = []
     details = ['<details>', '<summary>DDL, index configuration and image provenance</summary>', '']
-    raw_bytes = read(root / 'corpus' / 'size.json').get('raw_jsonl_bytes')
+    raw_bytes = read(data_root / 'corpus' / 'size.json').get('raw_jsonl_bytes')
     for t in targets:
         directory = result_dir(root, t)
         physical = read(directory / 'physical-evidence.json')
         storage = physical.get('storage', {})
-        total = storage.get('engine_total_bytes')
+        s3 = read(root / f'{t}-1' / 'storage.json').get('type') == 's3'
+        total = None if s3 else storage.get('engine_total_bytes')
         stats = read(root / f'{t}-1' / 'post-load-docker-stats.json')
         ratio = f'{raw_bytes / total:.2f}×' if (physical.get('passed') is True and
                  isinstance(raw_bytes, (int, float)) and raw_bytes > 0 and
@@ -186,6 +188,7 @@ def lifecycle(root, targets):
     lines += ['## Load and physical evidence', '', table(['DB', 'Write transport', 'Batches', 'Load smoke rows'], load_rows), '',
               f'Raw JSONL size: **{size(raw_bytes)}**. Ratio = raw JSONL / engine bytes at collection time; '
               'not a settled-storage guarantee. CPU/memory are post-load snapshots, not peaks. '
+              'S3 total/compression is omitted: local files exclude remote objects. '
               'Missing artifacts remain blank; pre-cleanup stats are not substituted.', '',
               table(['DB', 'Engine size', 'Size ratio', 'Materialization', 'SSTs / active parts',
                      'Post-load CPU', 'Post-load memory / limit'], physical_rows), '', *details, '</details>', '']

.github/scripts/aliyun-ecs-runner-provision.py

@@ -59,6 +59,7 @@ MANAGED_BY_TAG_KEY = "managed-by"
 MANAGED_BY_TAG_VALUE = "query-regression-ci"
 RUN_TAG_KEY = "query-regression-run-id"
 TTL_TAG_KEY = "runner-ttl-hours"
+REPO_TAG_KEY = "github-repository"
 
 # Runner cache paths on the instance system disk. They are created empty
 # every provision and deleted with the VM.
@@ -359,6 +360,7 @@ def run_instance(client, config: ProvisionConfig, user_data: str) -> str:
                 key=MANAGED_BY_TAG_KEY, value=MANAGED_BY_TAG_VALUE
             ),
             ecs_models.RunInstancesRequestTag(key=RUN_TAG_KEY, value=config.run_id),
+            ecs_models.RunInstancesRequestTag(key=REPO_TAG_KEY, value=config.repo),
         ]
         + (
             [

.github/scripts/aliyun-ecs-runner-teardown.py

@@ -32,8 +32,8 @@ Two modes:
   runner by name. Both steps are idempotent and best-effort; a missing
   instance or runner is not an error.
 - Sweep (scheduled janitor): delete every instance tagged as managed by
-  query-regression CI whose creation time is older than its tagged TTL (or
-  the given fallback TTL for untagged instances), and
+  query-regression CI in the requested repository whose creation time is older
+  than its tagged TTL (or the fallback TTL for instances without a TTL tag), and
   deregister the matching runners. This is the safety net for runs whose
   teardown job never executed.
 """
@@ -162,7 +162,7 @@ def delete_instance(client, instance_id: str, region_id: str | None = None) -> b
 
 
 def list_managed_instances(
-    client, region_id: str
+    client, region_id: str, repo: str
 ) -> list[tuple[str, str, str, str | None]]:
     from alibabacloud_ecs20140526 import models as ecs_models
 
@@ -175,6 +175,9 @@ def list_managed_instances(
                 ecs_models.DescribeInstancesRequestTag(
                     key=provision.MANAGED_BY_TAG_KEY,
                     value=provision.MANAGED_BY_TAG_VALUE,
+                ),
+                ecs_models.DescribeInstancesRequestTag(
+                    key=provision.REPO_TAG_KEY, value=repo
                 )
             ],
             max_results=100,
@@ -201,7 +204,7 @@ def list_managed_instances(
 
 
 def sweep(client, region_id: str, repo: str, github_token: str, ttl: timedelta) -> int:
-    instances = list_managed_instances(client, region_id)
+    instances = list_managed_instances(client, region_id, repo)
     print(f"Found {len(instances)} managed instance(s) in {region_id}", flush=True)
     expired = expired_instance_names(instances, datetime.now(timezone.utc), ttl)
     ok = True
@@ -227,7 +230,7 @@ def main() -> int:
     parser.add_argument(
         "--sweep",
         action="store_true",
-        help="Janitor mode: delete all managed instances older than --sweep-ttl-hours.",
+        help="Janitor mode: delete expired managed instances tagged for --repo.",
     )
     parser.add_argument("--sweep-ttl-hours", type=float, default=SWEEP_TTL.total_seconds() / 3600)
     args = parser.parse_args()

.github/scripts/benchmark-version-comparison.py

@@ -62,11 +62,12 @@ def row(query, metric, old, new, compatible, reason):
                 improvement_percent=delta, status='comparable' if delta is not None else reason)
 
 
-def agent(root):
+def agent(root, storage="file"):
+    candidate, baseline = (root, root / "baseline") if storage == "file" else (root / "s3", root / "baseline/s3")
     def load(r):
         return read(r / 'greptimedb-1/greptimedb-1/measured-summary.json')
-    a, b = load(root / 'baseline'), load(root)
-    finished = [r / 'greptimedb-1/exit-code.txt' for r in (root / 'baseline', root)]
+    a, b = load(baseline), load(candidate)
+    finished = [r / 'greptimedb-1/exit-code.txt' for r in (baseline, candidate)]
     compatible = (all(p.is_file() and p.read_text().strip() == '0' for p in finished)
                   and a.get('passed') is True and b.get('passed') is True
                   and same(a, b, ('contract_sha256', 'data_profile', 'query_window', 'warmup_runs', 'measured_runs')))
@@ -81,8 +82,9 @@ def agent(root):
     return rows
 
 
-def traces(root):
-    a, b = (read(r / 'runs/greptimedb/run.json') for r in (root / 'baseline', root))
+def traces(root, storage="file"):
+    candidate, baseline = (root, root / "baseline") if storage == "file" else (root / "s3", root / "baseline/s3")
+    a, b = (read(r / 'runs/greptimedb/run.json') for r in (baseline, candidate))
     compatible = (a.get('success') is True and b.get('success') is True
                   and same(a, b, ('expected', 'protocol', 'audit_expected', 'corpus'))
                   and same(a.get('execution', {}), b.get('execution', {}),
@@ -110,18 +112,22 @@ def traces(root):
     return rows
 
 
-def long_range(root):
+def long_range(root, storage="file"):
+    names = ("greptimedb-baseline", "greptimedb") if storage == "file" else ("greptimedb-s3-baseline", "greptimedb-s3")
+    def target_root(name):
+        nested = root / "o11ybench" / ("o11ybench-" + name)
+        return nested if nested.exists() else root
     rows = []
-    manifest = read(root / 'run-manifest.json')
+    manifests = {name: read(target_root(name) / 'run-manifest.json') for name in names}
     healthy = all(
-        all(field in (manifest.get(phase, {}).get(target, {}).get('docker', {}) if phase == 'post_load_results'
-                      else manifest.get(phase, {}).get(target, {})).get('state', '').split()
+        all(field in (manifests[target].get(phase, {}).get(target, {}).get('docker', {}) if phase == 'post_load_results'
+                      else manifests[target].get(phase, {}).get(target, {})).get('state', '').split()
             for field in ('Running=true', 'ExitCode=0', 'OOMKilled=false'))
         for phase in ('post_load_results', 'post_query_results')
-        for target in ('greptimedb-baseline', 'greptimedb'))
+        for target in names)
     for mode in ('serial', 'concurrent-duration-c1', 'concurrent-duration-c4'):
-        a, b = (read(root / name / 'results-vmbench-long-range' / (mode + '-summary.json'))
-                for name in ('greptimedb-baseline', 'greptimedb'))
+        a, b = (read(target_root(name) / name / 'results-vmbench-long-range' / (mode + '-summary.json'))
+                for name in names)
         fields = ('workload', 'query_file_sha256', 'start', 'end', 'step', 'profile', 'data_duration',
                   'scrape_interval', 'cache_policy_effective', 'mode', 'connection_mode', 'request_timeout_seconds')
         fields += ('tries', 'warmup_runs') if mode == 'serial' else ('concurrency', 'duration_seconds', 'warmup_seconds')
@@ -178,7 +184,13 @@ def main():
         parser.error('report requires --root and --kind')
     document = {'kind': args.kind, 'baseline_tag': os.environ.get('BASELINE_GREPTIMEDB_TAG', ''),
                 'candidate_tag': os.environ.get('GREPTIMEDB_TAG', ''),
-                'rows': {'agent': agent, 'traces': traces, 'long-range': long_range}[args.kind](args.root)}
+                'rows': []}
+    mode = os.environ.get('GREPTIMEDB_STORAGE', 'file')
+    for storage in (['file', 's3'] if mode == 'both' else [mode]):
+        rows = {'agent': agent, 'traces': traces, 'long-range': long_range}[args.kind](args.root, storage)
+        for item in rows:
+            item['query'] = storage.upper() + '/' + item['query']
+        document['rows'].extend(rows)
     text = render(document)
     # Preserve partial comparisons even if a target failed before creating output.
     args.root.mkdir(parents=True, exist_ok=True)

.github/scripts/build-query-regression-helpers.sh

@@ -0,0 +1,43 @@
+#!/usr/bin/env bash
+set -euo pipefail
+: "${CARGO_TARGET_DIR:?CARGO_TARGET_DIR must be configured by the runner}"
+revision="$(git rev-parse HEAD)"
+tag="query-regression-helpers:${revision}"
+staging="$(mktemp -d)"
+trap 'rm -rf -- "$staging"' EXIT
+cargo build --profile nightly -p cmd --features dev-tools \
+  --bin query_perf_fixture --bin query_regression_runner
+for binary in query_perf_fixture query_regression_runner; do
+  "$CARGO_TARGET_DIR/nightly/$binary" --help >/dev/null
+  cp "$CARGO_TARGET_DIR/nightly/$binary" "$staging/$binary"
+done
+mkdir -p helpers-image helpers-validation
+docker build --provenance=false --build-arg "SOURCE_REVISION=$revision" \
+  --file docker/query-regression-helpers/Dockerfile --tag "$tag" "$staging"
+container="$(docker create "$tag" /query_perf_fixture --help)"
+trap 'docker rm --force --volumes "$container"; rm -rf -- "$staging"' EXIT
+mkdir -p "$staging/extracted"
+for binary in query_perf_fixture query_regression_runner; do
+  docker cp "$container:/$binary" "$staging/extracted/$binary"
+  chmod +x "$staging/extracted/$binary"
+  "$staging/extracted/$binary" --help >/dev/null
+  cmp "$staging/$binary" "$staging/extracted/$binary"
+done
+# Hold the revision and cases constant: compare source-built versus extracted
+# fixture normalization twice. Keep the raw outputs as workflow artifacts.
+printf '%s\n' "$revision" > helpers-validation/source-revision.txt
+for iteration in 1 2; do
+  while IFS= read -r case; do
+    name="$(basename "$(dirname "$case")")"
+    for mode in source image; do
+      binary="$staging/query_perf_fixture"
+      [[ "$mode" != image ]] || binary="$staging/extracted/query_perf_fixture"
+      "$binary" plan --case "$case" > "helpers-validation/${name}-${iteration}-${mode}.json"
+    done
+    cmp "helpers-validation/${name}-${iteration}-source.json" \
+        "helpers-validation/${name}-${iteration}-image.json"
+  done < <(find tests/perf/query_cases -name case.toml -type f | sort)
+done
+(cd "$staging" && sha256sum query_perf_fixture query_regression_runner) > helpers-image/SHA256SUMS
+printf '%s\n' "$revision" > helpers-image/source-revision.txt
+docker save "$tag" --output helpers-image/image.tar

.github/scripts/ci-runners.md

@@ -13,8 +13,12 @@ AWS uses `aws-ec2-runner-provision.py` and `aws-ec2-runner-teardown.py` directly
 Repository variables supply `EC2_RUNNER_REGION`, `EC2_RUNNER_SUBNET_ID`, and
 `EC2_RUNNER_SECURITY_GROUP_ID`; credentials use `AWS_ACCESS_KEY_ID`,
 `AWS_SECRET_ACCESS_KEY`, and `GH_PERSONAL_ACCESS_TOKEN` secrets.
-The current region is Singapore (`ap-southeast-1`). By default the script selects
-the newest available Canonical Ubuntu 24.04 amd64 server AMI in that region
+The runner region comes from `EC2_RUNNER_REGION` (currently `us-east-1`).
+The subnet and security group must belong to the selected AWS account and region.
+For benchmarks using real S3, `AWS_CI_TEST_BUCKET_REGION` must match the runner
+region; `AWS_CI_TEST_BUCKET` must be accessible with the configured credentials.
+By default the script selects the newest available Canonical Ubuntu 24.04 amd64
+server AMI in that region
 (owner `099720109477`) and logs its ID. Set `BENCHMARK_EC2_IMAGE_ID` to pin a
 prepared Ubuntu AMI. Existing build CI's `EC2_RUNNER_LINUX_AMD64_IMAGE_ID` is
 unaffected. Both workflows default `system_disk_gib` to `auto` (also accepting
@@ -31,9 +35,12 @@ swappiness 10, disables systemd-oomd and background automatic updates, and sets
 `OOMPolicy=continue` on the runner service. Explicit setup package installation
 remains enabled.
 
-Both benchmark jobs call `.github/actions/setup-benchmark` after checkout. It
-reuses existing tools, installs missing curl/jq/python3/rsync, installs Docker CE
-if absent, and checks Docker access. Missing packages require passwordless sudo;
+Benchmark jobs call `.github/actions/setup-linux-runner` after checkout with
+`tools: curl jq python3 rsync` and require x86_64. Linux artifact builds reuse
+the same action with `tools: make shasum` on either amd64 or arm64; the RISC-V
+cross-build calls it on amd64 too. The action reuses installed commands, maps
+`shasum` to Ubuntu package `libdigest-sha-perl`, installs Docker CE if absent,
+and checks Docker access. Missing packages require passwordless sudo;
 Aliyun's prepared image normally provides the tools and Docker already. Existing
 Docker installations are preserved, so exact versions need not match between
 providers. Provisioning must establish Docker permissions before starting the

.github/scripts/metric-snapshot-s3-bench.py

@@ -105,8 +105,9 @@ class Benchmark:
         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")
+        if not local_smoke and (not os.environ["S3_REGION"].strip()
+                                or os.environ["S3_REGION"] != os.environ["RUNNER_REGION"]):
+            raise ValueError("Cloud benchmark requires the same nonempty bucket and runner region")
         self.containers = set()
         self.network_created = False
         self.phases = []
@@ -324,7 +325,7 @@ level = "info"
         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.", "",
+                 f"S3 in {os.environ['S3_REGION']}; 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']} |")

.github/scripts/metric-snapshot-s3-ops.py

@@ -21,7 +21,7 @@
 """Budget preflight and strictly scoped S3 operations for the metric snapshot benchmark."""
 
 import argparse
-from decimal import Decimal
+from decimal import Decimal, InvalidOperation
 import json
 import os
 from pathlib import Path
@@ -29,8 +29,8 @@ import re
 import time
 import urllib.request
 
-REGION = "ap-southeast-1"
-HOURLY_USD = Decimal("0.9408")
+# Historical budget estimate; dispatch can supply a rate for the configured region.
+DEFAULT_HOURLY_USD = "0.9408"
 OTHER_RESERVE_USD = Decimal("1.50")
 
 
@@ -40,6 +40,16 @@ def owned_prefix(prefix):
     return prefix
 
 
+def hourly_estimate(env):
+    try:
+        estimate = Decimal(env.get("EC2_HOURLY_USD", DEFAULT_HOURLY_USD))
+    except InvalidOperation as error:
+        raise ValueError("EC2_HOURLY_USD must be a positive finite estimate") from error
+    if not estimate.is_finite() or estimate <= 0:
+        raise ValueError("EC2_HOURLY_USD must be a positive finite estimate")
+    return estimate
+
+
 def plan(env):
     image = env["GREPTIMEDB_IMAGE"]
     if not re.fullmatch(r"greptime/greptimedb(?:-nightly)?@sha256:[0-9a-f]{64}", image):
@@ -49,21 +59,22 @@ def plan(env):
     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_REGION"].strip() or env["S3_REGION"] != env["RUNNER_REGION"]:
+        raise ValueError("bucket and runner must use the same nonempty region")
     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
+    hourly = hourly_estimate(env)
+    reservation = hourly * 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,
+            "instance_type": "c6id.4xlarge", "region": env["S3_REGION"],
             "prior_estimated_usd": str(prior), "attempt_reserve_usd": str(reservation),
-            "ec2_hourly_usd": str(HOURLY_USD), "other_reserve_usd": str(OTHER_RESERVE_USD),
+            "ec2_hourly_usd": str(hourly), "other_reserve_usd": str(OTHER_RESERVE_USD),
             "cost_note": "Estimate, not an AWS billing total or a hard spending cap."}
 
 
@@ -71,8 +82,8 @@ def client(region):
     import boto3
     from botocore.config import Config
 
-    if region != REGION:
-        raise ValueError("S3 region must be ap-southeast-1")
+    if not region.strip():
+        raise ValueError("S3 region is required")
     return boto3.client("s3", region_name=region, config=Config(
         connect_timeout=10, read_timeout=30, retries={"mode": "standard", "max_attempts": 3}))
 
@@ -193,8 +204,10 @@ def main():
         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)),
+        hourly = hourly_estimate(os.environ)
+        estimate = hourly * Decimal(seconds) / 3600
+        result = {"allocation_window_seconds": seconds, "ec2_hourly_usd": str(hourly),
+                  "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."}

.github/scripts/o11ybench-storage.py

@@ -0,0 +1,316 @@
+# 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.
+
+"""Select File/S3 benchmark arms without changing workload adapters."""
+import argparse
+import copy
+import importlib.util
+import json
+from pathlib import Path
+import sys
+import os
+import subprocess
+
+
+def validate(env):
+    mode = env.get('GREPTIMEDB_STORAGE', 'file')
+    if mode not in ('file', 's3', 'both'):
+        raise ValueError('GREPTIMEDB_STORAGE must be file, s3 or both')
+    if mode != 'file':
+        if env.get('RUN_GREPTIMEDB') != 'true':
+            raise ValueError('S3 requires Run GreptimeDB')
+        if env.get('PROVIDER') != 'AWS':
+            raise ValueError('S3 comparisons require an AWS runner')
+        if not env.get('S3_BUCKET') or not env.get('S3_REGION'):
+            raise ValueError('AWS_CI_TEST_BUCKET and AWS_CI_TEST_BUCKET_REGION must be configured')
+        if env['S3_REGION'] != env.get('RUNNER_REGION'):
+            raise ValueError('S3 bucket and EC2 runner regions must match')
+    return mode
+
+
+def replace(command, key, value):
+    result = list(command)
+    if key in result:
+        result[result.index(key) + 1] = value
+    else:
+        result += [key, value]
+    return result
+
+
+def commands(command, mode):
+    """Recovery visits each owned root; execution uses explicit workflow steps."""
+    stage = command[command.index('--stage') + 1]
+    if stage != 'cleanup':
+        raise ValueError('only cleanup may expand multiple arms')
+    result = [list(command)]
+    if mode != 'file':
+        root = command[command.index('--root') + 1]
+        owner = command[command.index('--owner') + 1]
+        secondary = replace(command, '--root', root + '/s3')
+        result.append(replace(secondary, '--owner', owner + '-s3'))
+    return result
+
+
+def module(path):
+    sys.path.insert(0, str(path.parent))
+    spec = importlib.util.spec_from_file_location(path.stem.replace('-', '_'), path)
+    result = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(result)
+    return result
+
+
+def summary(kind, root, targets, section, mode):
+    filename = 'agent-observability-summary.py' if kind == 'agent' else 'tracesbench-summary.py'
+    renderer = module(Path(__file__).with_name(filename))
+    primary = [t for t in targets if t != 'greptimedb' or mode != 's3']
+    arms = [(root, primary, 'File')]
+    if mode != 'file' and 'greptimedb' in targets:
+        arms.append((root / 's3', ['greptimedb'], 'S3'))
+    output = []
+    for arm, selected, label in arms:
+        if not selected:
+            continue
+        output.append('## ' + ('GreptimeDB S3' if label == 'S3' else 'Primary targets (GreptimeDB File)'))
+        if kind == 'agent':
+            body = (renderer.benchmark(arm, selected) if section == 'benchmark'
+                    else renderer.lifecycle(arm, selected, data_root=root.parent if root.name == 'baseline' else root))
+        else:
+            report, runs = renderer.load(arm)
+            body = (renderer.benchmark(report, runs, selected) if section == 'benchmark'
+                    else renderer.lifecycle(report, runs, arm, selected))
+        body = body.replace('| greptimedb |', '| GreptimeDB ' + label + ' |')
+        body = body.replace('**greptimedb**', '**GreptimeDB ' + label + '**')
+        output.append(body)
+        if label == 'S3':
+            directories = [arm / 'greptimedb-1'] if kind == 'agent' else [arm / 'runs/greptimedb']
+            for directory in directories:
+                for action in ('probe', 'inventory', 'cleanup'):
+                    path = directory / ('s3-' + action + '.json')
+                    data = json.loads(path.read_text()) if path.exists() else {}
+                    status = 'passed' if data.get('passed') is True else 'failed / unavailable'
+                    count = data.get('object_count', data.get('deleted_objects', '—'))
+                    output.append(f'- S3 {action}: **{status}**; objects: {count}.')
+    return '\n\n'.join(output)
+
+
+def split_metrics(plan):
+    """One executor root per target, with the immutable dataset shared by all."""
+    result = {}
+    for target in plan['targets']:
+        child = copy.deepcopy(plan)
+        name = target['name']
+        dest = str(Path(plan['output_root']) / 'o11ybench' / ('o11ybench-' + name))
+        child['output_root'] = dest
+        child['run_id'] = plan['run_id'] + '-' + name
+        child['targets'] = [copy.deepcopy(target)]
+        child['data']['generate'] = False
+        child['report'].update(root=dest + '/report', index_html=dest + '/report/index.html')
+        child['artifacts'] = {k: dest + '/' + Path(v).name for k, v in plan['artifacts'].items()}
+        result[name] = child
+    return result
+
+
+def metrics(action, args):
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--plan', type=Path, required=True)
+    parser.add_argument('--tools', type=Path, required=True)
+    parser.add_argument('--runtime-image', required=True)
+    parser.add_argument('--target')
+    a = parser.parse_args(args)
+    plan = json.loads(a.plan.read_text())
+    plan['runtime_image'] = a.runtime_image
+    executor = module(a.tools / 'scripts/run_benchmark_plan.py')
+    root = Path(plan['output_root'])
+    if action == 'metrics-prepare':
+        executor.require_safe_execute_root(root)
+        root.mkdir(parents=True, exist_ok=False)
+        manifest = executor.build_manifest(plan)
+        executor.run_checked([str(executor.DOWNLOAD_DATA)], manifest,
+                             env=executor.generation_environment(plan))
+        (root / 'generation.json').write_text(json.dumps(manifest, indent=2) + '\n')
+        for name, child in split_metrics(plan).items():
+            (a.plan.parent / (name + '.json')).write_text(json.dumps(child, indent=2) + '\n')
+    elif action == 'metrics-run':
+        path = a.plan.parent / (a.target + '.json')
+        child = json.loads(path.read_text())
+        executor.execute_plan(child, executor.build_manifest(child), replace=False)
+        subprocess.run([sys.executable, str(a.tools / 'scripts/check_run_benchmark_plan.py'),
+                        '--artifacts', child['output_root']], check=True)
+    else:
+        # Validate every selected target together, retaining cross-target gates.
+        manifests = [executor.build_manifest(child) for child in split_metrics(plan).values()]
+        combined = executor.build_manifest(plan)
+        combined['targets'] = [target for m in manifests for target in m['targets']]
+        for command in executor.validation_commands(combined):
+            subprocess.run(command, check=True)
+
+
+def overview(path):
+    """Render the same validated report data into the GitHub job summary."""
+    data = json.loads(path.read_text())
+    renderer = module(Path(__file__).with_name('agent-observability-summary.py'))
+    lines = ['# Storage comparison', '', 'Smoke results are diagnostic; see classification per arm. '
+             'No cold-object-store performance claim. Missing results are not passes.', '']
+    for workload in data['workloads']:
+        labels = {p['id']: p['label'] for p in workload['participants']}
+        for scenario in workload['scenarios']:
+            lines += ['## ' + workload['title'] + ' / ' + scenario['label'], '']
+            rows = []
+            for result in scenario['results']:
+                lines += [labels[result['participant']] + ': ' + str(result.get('classification')), '']
+                for query_id, query in result['queries'].items():
+                    m = query['metrics']
+                    rows.append([query_id, labels[result['participant']], query['status'],
+                                 renderer.duration(m.get('p50_ms', m.get('median_ms'))),
+                                 renderer.duration(m.get('p95_ms')), renderer.duration(m.get('p99_ms'))])
+            lines.append(renderer.table(['Query', 'DB / storage', 'Status', 'P50 / median', 'P95', 'P99'], rows))
+    return '\n'.join(lines)
+
+
+def audit(root, report, targets, mode, baseline=False, allow_trace_query_failures=False):
+    data = json.loads(report.read_text())
+    actual = {(p['target'], p.get('storage', {}).get('type'), p['role'])
+              for w in data['workloads'] for p in w['participants']}
+    expected = {(t, None, 'comparison') for t in targets if t != 'greptimedb'}
+    modes = ['file', 's3'] if mode == 'both' else [mode]
+    if 'greptimedb' in targets:
+        expected |= {('greptimedb', m, role) for m in modes
+                     for role in (['candidate', 'baseline'] if baseline else ['candidate'])}
+    if actual != expected:
+        raise ValueError(f'participant mismatch: expected {sorted(map(str, expected))}, got {sorted(map(str, actual))}')
+    query_failures = []
+    for w in data['workloads']:
+        partial = allow_trace_query_failures and w.get('id') == 'traces'
+        if partial and not w['scenarios']:
+            raise ValueError('missing trace scenarios')
+        for scenario in w['scenarios']:
+            if partial and {r['participant'] for r in scenario['results']} != {p['id'] for p in w['participants']}:
+                raise ValueError('missing trace participant results')
+            for result in scenario['results']:
+                if partial and result.get('health') != 'recorded healthy snapshots':
+                    raise ValueError('missing or unhealthy trace evidence: ' + result['participant'])
+                queries = result['queries']
+                if not any(q['status'] == 'passed' for q in queries.values()):
+                    raise ValueError('no successful measured queries: ' + result['participant'])
+                if any(q['status'] == 'unavailable' or (q['status'] == 'failed' and not partial)
+                       for q in queries.values()):
+                    raise ValueError('failed or unavailable queries: ' + result['participant'])
+                for query_id, query in queries.items():
+                    if query['status'] == 'failed':
+                        query_failures.append(dict(workload=w['id'], scenario=scenario['id'],
+                                                   participant=result['participant'], query=query_id,
+                                                   reason=query.get('reason', '')))
+    for manifest_path in root.glob('o11ybench/*/run-manifest.json'):
+        manifest = json.loads(manifest_path.read_text())
+        if manifest.get('status') != 'passed':
+            raise ValueError('incomplete target manifest: ' + str(manifest_path))
+    probes = list(root.rglob('s3-probe.json'))
+    expected_s3 = sum(m == 's3' for _, m, _ in expected)
+    if len(probes) != expected_s3:
+        raise ValueError('missing or duplicate S3 evidence')
+    for probe in probes:
+        for action in ('probe', 'inventory', 'cleanup'):
+            payload = json.loads(probe.with_name('s3-' + action + '.json').read_text())
+            if payload.get('passed') is not True or (action == 'inventory' and payload.get('object_count', 0) <= 0):
+                raise ValueError('failed S3 ' + action)
+    text = (root / 'ci-summary.md').read_text()
+    for m in modes if 'greptimedb' in targets else []:
+        if m.upper() not in text:
+            raise ValueError('storage mode missing from saved summary: ' + m)
+    result = dict(passed=True, participants=[dict(target=t, storage=m, role=role)
+                  for t, m, role in sorted(expected, key=str)], s3_arms=len(probes), query_failures=query_failures)
+    (root / 'artifact-audit.json').write_text(json.dumps(result, indent=2) + '\n')
+    lines = ['## Artifact validation', '', '- Selected participants: ' + str(len(expected)) + '; all present.',
+             '- Query result coverage and S3 probe/inventory/cleanup: passed.', '']
+    if query_failures:
+        lines += ['- Collection completed with query failures; this is not an all-pass query run.', '']
+        lines += [f"- {f['participant']} / {f['query']}: failed (excluded from comparison)."
+                  for f in query_failures]
+        lines.append('')
+    renderer = module(Path(__file__).with_name('agent-observability-summary.py'))
+    rows = []
+    for probe in probes:
+        inv = json.loads(probe.with_name('s3-inventory.json').read_text())
+        cleanup = json.loads(probe.with_name('s3-cleanup.json').read_text())
+        rows.append([str(probe.parent.relative_to(root)), inv['object_count'], inv['object_bytes'],
+                     cleanup.get('deleted_objects'), 'passed'])
+    if rows:
+        lines.append(renderer.table(['S3 arm', 'Objects', 'Remote bytes', 'Deleted current objects', 'Cleanup'], rows))
+    text = '\n'.join(lines) + '\n'
+    with (root / 'ci-summary.md').open('a') as output:
+        output.write(text)
+    if os.environ.get('GITHUB_STEP_SUMMARY'):
+        with Path(os.environ['GITHUB_STEP_SUMMARY']).open('a') as output:
+            output.write(text)
+    return result
+
+
+def main():
+    if len(sys.argv) > 1 and sys.argv[1] == 'audit':
+        parser = argparse.ArgumentParser()
+        parser.add_argument('--root', type=Path, required=True)
+        parser.add_argument('--report', type=Path, required=True)
+        parser.add_argument('--targets', required=True)
+        parser.add_argument('--allow-trace-query-failures', action='store_true',
+                            help='Retain failed Traces queries after native validation; missing results still fail')
+        args = parser.parse_args(sys.argv[2:])
+        print(json.dumps(audit(args.root, args.report, args.targets.split(','),
+                              os.environ.get('GREPTIMEDB_STORAGE', 'file'),
+                              os.environ.get('COMPARE_GREPTIMEDB') == 'true', args.allow_trace_query_failures)))
+        return
+    if len(sys.argv) > 1 and sys.argv[1] == 'normalize-metrics':
+        path = Path(sys.argv[2])
+        plan = json.loads(path.read_text())
+        for target in plan['targets']:
+            if target.get('storage', {}).get('type') == 's3' and '-s3' not in target['name']:
+                target['name'] = target['name'].replace('-baseline', '-s3-baseline') if target['name'].endswith('-baseline') else target['name'] + '-s3'
+        path.write_text(json.dumps(plan, indent=2) + '\n')
+        return
+    if len(sys.argv) > 1 and sys.argv[1].startswith('metrics-'):
+        metrics(sys.argv[1], sys.argv[2:])
+        return
+    if len(sys.argv) > 1 and sys.argv[1] == 'overview':
+        print(overview(Path(sys.argv[2])))
+        return
+    if len(sys.argv) > 1 and sys.argv[1] == 'summary':
+        parser = argparse.ArgumentParser()
+        parser.add_argument('--kind', choices=['agent', 'traces'], required=True)
+        parser.add_argument('--root', type=Path, required=True)
+        parser.add_argument('--targets', required=True)
+        parser.add_argument('--section', choices=['benchmark', 'lifecycle'], required=True)
+        args = parser.parse_args(sys.argv[2:])
+        print(summary(args.kind, args.root, args.targets.split(','), args.section,
+                      os.environ.get('GREPTIMEDB_STORAGE', 'file')))
+        return
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('action', choices=['validate', 'run'])
+    parser.add_argument('command', nargs=argparse.REMAINDER)
+    args = parser.parse_args()
+    mode = validate(os.environ)
+    if args.action == 'validate':
+        print('GreptimeDB storage: ' + mode)
+        return
+    command = args.command[1:] if args.command[:1] == ['--'] else args.command
+    planned = commands(command, mode)
+    failed = 0
+    for cmd in planned:
+        result = subprocess.run(cmd)
+        failed = failed or result.returncode
+        # Cleanup must try both arms even if one failed. Target failures remain
+        # visible, while the other arm can still produce diagnostic evidence.
+    raise SystemExit(failed)
+
+
+if __name__ == '__main__':
+    main()

.github/scripts/prepare-fuzz-corpus.sh

@@ -0,0 +1,38 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+: "${FUZZ_TARGETS:?FUZZ_TARGETS is required}"
+: "${GITHUB_WORKSPACE:?GITHUB_WORKSPACE is required}"
+: "${GITHUB_OUTPUT:?GITHUB_OUTPUT is required}"
+
+targets=()
+while IFS= read -r target; do
+  target="${target%$'\r'}"
+  [[ -z "${target}" ]] && continue
+  if [[ ! "${target}" =~ ^[a-zA-Z0-9_]+$ ]]; then
+    printf 'Invalid fuzz target: %s\n' "${target}" >&2
+    exit 2
+  fi
+  targets+=("${target}")
+done <<<"${FUZZ_TARGETS}"
+
+if [[ "${#targets[@]}" -eq 0 ]]; then
+  printf 'FUZZ_TARGETS contains no targets\n' >&2
+  exit 2
+fi
+
+# Corpus identity depends on the targets, not the source revision or their order.
+target_set="$(printf '%s\n' "${targets[@]}" | LC_ALL=C sort -u)"
+target_hash="$(printf '%s\n' "${target_set}" | sha256sum | cut -d ' ' -f 1)"
+corpus_root="${GITHUB_WORKSPACE}/tests-fuzz/corpus"
+
+{
+  printf 'key-prefix=fuzz-corpus-v1-%s-\n' "${target_hash}"
+  printf 'paths<<FUZZ_CORPUS_PATHS\n'
+  while IFS= read -r target; do
+    mkdir -p "${corpus_root}/${target}"
+    printf '%s\n' "${corpus_root}/${target}"
+  done <<<"${target_set}"
+  printf 'FUZZ_CORPUS_PATHS\n'
+} >>"${GITHUB_OUTPUT}"

.github/scripts/publish-query-regression-helpers.sh

@@ -0,0 +1,28 @@
+#!/usr/bin/env bash
+set -euo pipefail
+repository=greptime-registry.cn-hangzhou.cr.aliyuncs.com/tools/query-regression-helpers
+revision="$(cat helpers-image/source-revision.txt)"
+[[ "$revision" == "$GITHUB_SHA" ]]
+tag="${repository}:${revision}-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
+# Match release-dev-builder-images.yaml: publish ACR images with Skopeo from
+# a hosted runner. A failed publish can reuse the completed build artifact.
+mkdir -p helpers-validation
+docker run --rm \
+  -v "${DOCKER_CONFIG:-$HOME/.docker}:/root/.docker:ro" \
+  -v "$PWD/helpers-image:/image:ro" \
+  -v "$PWD/helpers-validation:/result" \
+  -e REGISTRY_AUTH_FILE=/root/.docker/config.json \
+  quay.io/skopeo/stable:latest --command-timeout 20m copy --retry-times 3 \
+  --digestfile /result/digest.txt \
+  docker-archive:/image/image.tar "docker://$tag"
+digest="$(cat helpers-validation/digest.txt)"
+[[ "$digest" =~ ^sha256:[0-9a-f]{64}$ ]]
+image="$repository@$digest"
+# The ECS build tested these exact bytes. Verify them again from the registry;
+# hosted runners need not share the ECS runtime library versions.
+python3 .github/scripts/query-regression-helpers.py "$image" helpers-validation/extracted \
+  --checksums helpers-image/SHA256SUMS
+jq -e --arg revision "$revision" '.source_revision == $revision' \
+  helpers-validation/extracted/helpers-image.json >/dev/null
+printf 'image=%s\n' "$image" >> "$GITHUB_OUTPUT"
+printf 'Published helpers: `%s` (source `%s`)\n' "$image" "$revision" >> "$GITHUB_STEP_SUMMARY"

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

@@ -0,0 +1,82 @@
+#!/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.
+
+"""Extract digest-pinned query regression tools from an artifact image."""
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import subprocess
+from pathlib import Path
+
+BINARIES = ("query_perf_fixture", "query_regression_runner")
+
+
+def resolve_image(value):
+    if value == "pinned":
+        value = os.environ.get("QUERY_REGRESSION_HELPERS_IMAGE", "")
+    if not re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9._/:@-]*@sha256:[0-9a-f]{64}", value):
+        raise ValueError("helpers_image requires an image@sha256 digest; set QUERY_REGRESSION_HELPERS_IMAGE by publishing a verified image before using 'pinned'")
+    return value
+
+
+def run(*args):
+    return subprocess.check_output(args, text=True).strip()
+
+
+def extract(image, destination, checksums=None):
+    expected = None
+    if checksums is not None:
+        entries = [line.split() for line in checksums.read_text().splitlines()]
+        expected = {name: digest for digest, name in entries}
+        if set(expected) != set(BINARIES) or len(entries) != len(BINARIES):
+            raise ValueError("Checksums must contain exactly the two helper binaries")
+    image = resolve_image(image)
+    destination.mkdir(parents=True, exist_ok=True)
+    run("docker", "pull", image)
+    metadata = json.loads(run("docker", "image", "inspect", image))[0]
+    revision = (metadata.get("Config", {}).get("Labels") or {}).get("org.opencontainers.image.revision", "")
+    if not re.fullmatch(r"[0-9a-f]{40}", revision):
+        raise ValueError("Helper image is missing its source revision label")
+    container = run("docker", "create", image, "/query_perf_fixture", "--help")
+    try:
+        for binary in BINARIES:
+            target = destination / binary
+            run("docker", "cp", f"{container}:/{binary}", str(target))
+            target.chmod(0o755)
+            if expected is None:
+                run(str(target.resolve()), "--help")
+            else:
+                with target.open("rb") as file:
+                    actual = hashlib.file_digest(file, "sha256").hexdigest()
+                if actual != expected[binary]:
+                    raise ValueError(f"Published helper checksum mismatch: {binary}")
+    finally:
+        run("docker", "rm", "--force", "--volumes", container)
+    provenance = {"image": image, "image_id": metadata["Id"], "source_revision": revision}
+    (destination / "helpers-image.json").write_text(json.dumps(provenance, indent=2) + "\n")
+    print(json.dumps(provenance))
+    return provenance
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("image")
+    parser.add_argument("destination", type=Path)
+    parser.add_argument("--checksums", type=Path, help="Verify build checksums instead of executing on this host")
+    args = parser.parse_args()
+    extract(args.image, args.destination, args.checksums)

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

@@ -21,10 +21,13 @@ import argparse
 import json
 import os
 import re
+import shlex
 import signal
 import socket
 import subprocess
+import sys
 import time
+import uuid
 import urllib.request
 from dataclasses import dataclass
 from pathlib import Path
@@ -130,6 +133,7 @@ class RunTarget:
     datanode_http_port: int
     datanode_data_dir: Path
     frontend_config: Path | None = None
+    containerized: bool = False
 
 
 def allocate_ports(n: int) -> list[int]:
@@ -151,6 +155,7 @@ def make_target(
     root: Path,
     ports: list[int],
     frontend_config: Path | None = None,
+    containerized: bool = False,
 ) -> RunTarget:
     work_dir = root / name
     return RunTarget(
@@ -167,6 +172,7 @@ def make_target(
         datanode_http_port=ports[3],
         datanode_data_dir=work_dir / "datanode-0" / "data",
         frontend_config=frontend_config,
+        containerized=containerized,
     )
 
 
@@ -258,7 +264,11 @@ def stop_component(
     procs: dict[tuple[str, str], subprocess.Popen[bytes]],
 ) -> None:
     proc = procs.pop((target.name, name), None)
-    if proc is None or proc.poll() is not None:
+    if proc is None:
+        return
+    if proc.poll() is not None:
+        if target.containerized and proc.returncode != 0:
+            raise RuntimeError(f"Docker target {target.name}/{name} exited with {proc.returncode}")
         return
     try:
         os.killpg(proc.pid, signal.SIGTERM)
@@ -272,6 +282,8 @@ def stop_component(
         except ProcessLookupError:
             return
         proc.wait(timeout=20)
+    if target.containerized and proc.returncode not in (0, 128 + signal.SIGTERM):
+        raise RuntimeError(f"Docker target {target.name}/{name} cleanup failed: {proc.returncode}")
 
 
 def restart_component(
@@ -284,9 +296,15 @@ def restart_component(
 
 
 def stop_all(targets: list[RunTarget], procs: dict[tuple[str, str], subprocess.Popen[bytes]]) -> None:
+    errors = []
     for target in reversed(targets):
         for name in ("frontend", "datanode", "metasrv"):
-            stop_component(target, name, procs)
+            try:
+                stop_component(target, name, procs)
+            except Exception as err:
+                errors.append(err)
+    if errors:
+        raise RuntimeError(f"component cleanup failed: {errors}") from errors[0]
 
 
 def load_plan(fixture_generator: Path, case_path: Path) -> dict[str, Any]:
@@ -313,8 +331,8 @@ def run_direct_case(
 ) -> int:
     ports = allocate_ports(16)
     targets = [
-        make_target("base", base_bin, work_dir, ports[:8]),
-        make_target("candidate", candidate_bin, work_dir, ports[8:]),
+        make_target("base", base_bin, work_dir, ports[:8], containerized=bool(getattr(args, "base_image", ""))),
+        make_target("candidate", candidate_bin, work_dir, ports[8:], containerized=bool(getattr(args, "candidate_image", ""))),
     ]
     for target in targets:
         if target.work_dir.exists() and any(target.work_dir.iterdir()):
@@ -407,6 +425,7 @@ def run_remote_case(
             work_dir,
             ports[:8],
             work_dir / "base" / "frontend-prom-store.toml",
+            containerized=bool(getattr(args, "base_image", "")),
         ),
         make_target(
             "candidate",
@@ -414,6 +433,7 @@ def run_remote_case(
             work_dir,
             ports[8:],
             work_dir / "candidate" / "frontend-prom-store.toml",
+            containerized=bool(getattr(args, "candidate_image", "")),
         ),
     ]
     for target in targets:
@@ -495,8 +515,8 @@ def run_otlp_case(
         raise ValueError("--otelgen-bin is required for otlp_trace_load")
     ports = allocate_ports(16)
     targets = [
-        make_target("base", base_bin, work_dir, ports[:8]),
-        make_target("candidate", candidate_bin, work_dir, ports[8:]),
+        make_target("base", base_bin, work_dir, ports[:8], containerized=bool(getattr(args, "base_image", ""))),
+        make_target("candidate", candidate_bin, work_dir, ports[8:], containerized=bool(getattr(args, "candidate_image", ""))),
     ]
     for target in targets:
         if target.work_dir.exists() and any(target.work_dir.iterdir()):
@@ -634,6 +654,106 @@ def write_failed_report(work_dir: Path, case_path: Path, error: Exception) -> No
     )
 
 
+def docker_target(image: str, root: Path, command: list[str]) -> int:
+    """Expose an image as a binary, including read-bench subcommands.
+
+    The caller terminates a process group. Keep Docker clients in separate
+    sessions so cleanup can stop the container before SST inspection starts.
+    """
+    name = f"query-regression-{uuid.uuid4().hex}"
+
+    def interrupted(signum, _frame):
+        raise SystemExit(128 + signum)
+
+    previous = {sig: signal.signal(sig, interrupted) for sig in (signal.SIGTERM, signal.SIGINT)}
+    try:
+        subprocess.run([
+            "docker", "create", "--name", name, "--network", "host",
+            "--user", f"{os.getuid()}:{os.getgid()}",
+            "--volume", f"{root}:{root}", "--workdir", str(root),
+            "--entrypoint", "/greptime/bin/greptime", image, *command,
+        ], check=True, stdout=subprocess.DEVNULL, start_new_session=True)
+        attached = subprocess.run(
+            ["docker", "start", "--attach", name], check=False, start_new_session=True,
+        )
+        if attached.returncode != 0:
+            return attached.returncode
+        # docker start's status is not the contract for the container's exit code.
+        result = subprocess.run(
+            ["docker", "inspect", "--format", "{{.State.ExitCode}}", name],
+            check=True, capture_output=True, text=True, start_new_session=True,
+        )
+        return int(result.stdout.strip())
+    finally:
+        for sig in previous:
+            signal.signal(sig, signal.SIG_IGN)
+        try:
+            subprocess.run(
+                ["docker", "stop", "--time", "5", name], check=False,
+                stdout=subprocess.DEVNULL, timeout=8, start_new_session=True,
+            )
+        except subprocess.TimeoutExpired:
+            # Force-removal below still waits for the container to stop.
+            pass
+        finally:
+            try:
+                subprocess.run(
+                    ["docker", "rm", "--force", "--volumes", name], check=True,
+                    stdout=subprocess.DEVNULL, timeout=5, start_new_session=True,
+                )
+            finally:
+                for sig, handler in previous.items():
+                    signal.signal(sig, handler)
+
+
+def prepare_targets(args: argparse.Namespace) -> dict[str, Any]:
+    """Resolve images once and create binary-compatible launchers for all cases."""
+    metadata = {}
+    images = {}
+    for name in ("base", "candidate"):
+        image = getattr(args, f"{name}_image")
+        ref = getattr(args, f"{name}_ref")
+        if not image:
+            metadata[name] = {"kind": "source", "ref": os.environ.get(f"VERIFIED_{name.upper()}_SHA") or ref}
+            continue
+        if sys.platform != "linux":
+            raise ValueError("Docker targets require Linux host networking and a local Docker daemon")
+        if image.startswith("-") or any(c.isspace() for c in image):
+            raise ValueError(f"invalid {name} image reference: {image!r}")
+        if image not in images:
+            result = subprocess.run(
+                ["docker", "image", "inspect", image], check=False, capture_output=True, text=True,
+            )
+            if result.returncode != 0:
+                subprocess.run(["docker", "pull", image], check=True)
+                result = subprocess.run(
+                    ["docker", "image", "inspect", image], check=True, capture_output=True, text=True,
+                )
+            images[image] = json.loads(result.stdout)[0]
+        info = images[image]
+        image_id = info["Id"]
+        if not re.fullmatch(r"sha256:[0-9a-f]{64}", image_id):
+            raise ValueError(f"invalid Docker image ID: {image_id!r}")
+        args.work_dir = args.work_dir.resolve()
+        if ":" in str(args.work_dir):
+            raise ValueError("Docker target work directory cannot contain ':'")
+        launchers = args.work_dir / ".targets"
+        launchers.mkdir(parents=True, exist_ok=True)
+        launcher = launchers / name
+        command = [sys.executable, str(Path(__file__).resolve()),
+                   "docker-target", image_id, str(args.work_dir)]
+        launcher.write_text("#!/bin/sh\nexec " + shlex.join(command) + ' "$@"\n', encoding="utf-8")
+        launcher.chmod(0o755)
+        setattr(args, f"{name}_bin", launcher)
+        setattr(args, f"{name}_ref", f"{image} ({image_id})")
+        metadata[name] = {
+            "kind": "docker", "image": image, "image_id": image_id,
+            "repo_digests": info.get("RepoDigests", []),
+            "network": "host", "user": f"{os.getuid()}:{os.getgid()}",
+        }
+    return metadata
+
+
 def main() -> int:
     parser = argparse.ArgumentParser()
     parser.add_argument("--cases", action="append", help="'all', 'heavy', or comma/space separated case paths")
@@ -663,6 +783,8 @@ def main() -> int:
     parser.add_argument("--base-ref", default=os.environ.get("BASE_REF", ""))
     parser.add_argument("--candidate-ref", default=os.environ.get("CANDIDATE_REF", ""))
     parser.add_argument("--github-output", default=os.environ.get("GITHUB_OUTPUT"))
+    parser.add_argument("--base-image", default=os.environ.get("BASE_IMAGE", ""))
+    parser.add_argument("--candidate-image", default=os.environ.get("CANDIDATE_IMAGE", ""))
     args = parser.parse_args()
     try:
         cases = split_cases(args.cases or [os.environ.get("CASE_PATHS", "all")])
@@ -671,6 +793,12 @@ def main() -> int:
         append_github_output(args.github_output, 1)
         return 0
 
+    preparation_error = None
+    metadata = {}
+    try:
+        metadata = prepare_targets(args)
+    except Exception as err:
+        preparation_error = err
     reports: list[Path] = []
     status = 0
     for case in cases:
@@ -678,11 +806,18 @@ def main() -> int:
         work_dir = args.work_dir / case_slug(case_path)
         reports.append(work_dir / "query-regression-report.json")
         try:
+            if preparation_error is not None:
+                raise preparation_error
             case_status = run_case(args, case_path, work_dir)
         except Exception as err:  # noqa: BLE001 - preserve an aggregate report for the summary step
             print(f"error: query regression case {case_path}: {err}", flush=True)
             write_failed_report(work_dir, case_path, err)
             case_status = 1
+        report_path = reports[-1]
+        if (args.base_image or args.candidate_image) and report_path.exists():
+            report = json.loads(report_path.read_text(encoding="utf-8"))
+            report["execution_targets"] = metadata
+            report_path.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
         if case_status != 0:
             status = case_status or 1
 
@@ -695,4 +830,6 @@ def main() -> int:
 
 
 if __name__ == "__main__":
+    if len(sys.argv) > 1 and sys.argv[1] == "docker-target":
+        raise SystemExit(docker_target(sys.argv[2], Path(sys.argv[3]), sys.argv[4:]))
     raise SystemExit(main())

.github/scripts/run-fuzz-targets-test.sh

@@ -5,6 +5,7 @@ set -euo pipefail
 SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
 RUNNER="${SCRIPT_DIR}/run-fuzz-targets.sh"
 COLLECTOR="${SCRIPT_DIR}/collect-fuzz-target-artifacts.sh"
+PREPARE_CORPUS="${SCRIPT_DIR}/prepare-fuzz-corpus.sh"
 
 fail() {
   printf 'FAIL: %s\n' "$*" >&2
@@ -32,6 +33,13 @@ new_fixture() {
 set -euo pipefail
 printf '%s\t%s\t%s\n' "${GT_FUZZ_DUMP_DIR}" "$*" "${MOCK_CARGO_MARKER:-}" >>"${MOCK_CARGO_LOG}"
 target="$4"
+corpus="$5"
+[[ -d "${corpus}" ]] || exit 18
+printf '%s\t%s\n' "${target}" "${corpus}" >>"${MOCK_CORPUS_LOG}"
+if [[ -f "${corpus}/seed" ]]; then
+  cat "${corpus}/seed" >>"${MOCK_SEED_LOG}"
+fi
+printf '%s\n' "${target}" >"${corpus}/learned"
 case " ${MOCK_FAIL_TARGETS:-} " in
   *" ${target} "*) exit 17 ;;
 esac
@@ -64,12 +72,15 @@ run_fixture() {
     "MOCK_COLLECTOR_LOG=${fixture}/collector.log"
     "MOCK_FAIL_TARGETS=${fail_targets}"
     "MOCK_CARGO_MARKER=fixture"
+    "MOCK_CORPUS_LOG=${fixture}/corpus.log"
+    "MOCK_SEED_LOG=${fixture}/seeds.log"
     "FUZZ_TARGETS=${targets}"
     "FUZZ_GROUP=test-group"
     "FUZZ_MAX_TOTAL_TIME=120"
     "FUZZ_UNSTABLE=${unstable}"
     "FUZZ_COLLECT_CLUSTER_ARTIFACTS=true"
     "FUZZ_ARTIFACT_ROOT=${artifact_root}"
+    "FUZZ_CORPUS_ROOT=${fixture}/corpus with spaces"
     "FUZZ_ARTIFACT_COLLECTOR=${fixture}/collector"
     "GT_FUZZ_BINARY_PATH=${GT_FUZZ_BINARY_PATH:-}"
     "GT_FUZZ_INSTANCE_ROOT_DIR=${GT_FUZZ_INSTANCE_ROOT_DIR:-}"
@@ -81,7 +92,7 @@ run_fixture() {
   fi
 
   set +e
-  env -u FUZZ_FAIL_FAST "${runner_env[@]}" \
+  env -u FUZZ_FAIL_FAST -u FUZZ_BIN_DIR "${runner_env[@]}" \
     "${RUNNER}" >"${fixture}/stdout.log" 2>&1
   fixture_status=$?
   set -e
@@ -89,18 +100,30 @@ run_fixture() {
 
 run_prebuilt_fixture() {
   new_fixture
-  mkdir -p "${fixture}/prebuilt"
+  mkdir -p "${fixture}/prebuilt" "${fixture}/corpus with spaces/fuzz_create_table"
+  printf 'restored prebuilt seed\n' >"${fixture}/corpus with spaces/fuzz_create_table/seed"
   cat >"${fixture}/prebuilt/fuzz_create_table" <<'EOF'
 #!/usr/bin/env bash
 set -euo pipefail
 printf '%s\t%s\t%s\n' "${GT_FUZZ_DUMP_DIR}" "$*" "${MOCK_BINARY_MARKER:-}" >>"${MOCK_BINARY_LOG}"
+[[ -d "$1" ]] || exit 18
+printf '%s\n' "$1" >>"${MOCK_CORPUS_LOG}"
+cat "$1/seed" >>"${MOCK_SEED_LOG}"
+printf 'new prebuilt input\n' >"$1/learned"
+exit "${MOCK_BINARY_EXIT_CODE}"
 EOF
   chmod +x "${fixture}/prebuilt/fuzz_create_table"
 
   set +e
   env \
+    "PATH=${fixture}/bin:${PATH}" \
+    "MOCK_CARGO_LOG=${fixture}/cargo.log" \
     "MOCK_BINARY_LOG=${fixture}/binary.log" \
     "MOCK_BINARY_MARKER=fixture" \
+    "MOCK_BINARY_EXIT_CODE=${1:-0}" \
+    "MOCK_CORPUS_LOG=${fixture}/corpus.log" \
+    "MOCK_SEED_LOG=${fixture}/seeds.log" \
+    "MOCK_COLLECTOR_LOG=${fixture}/collector.log" \
     "FUZZ_TARGETS=fuzz_create_table" \
     "FUZZ_GROUP=test-group" \
     "FUZZ_MAX_TOTAL_TIME=120" \
@@ -108,6 +131,7 @@ EOF
     "FUZZ_FAIL_FAST=true" \
     "FUZZ_COLLECT_CLUSTER_ARTIFACTS=false" \
     "FUZZ_ARTIFACT_ROOT=${fixture}/artifacts" \
+    "FUZZ_CORPUS_ROOT=${fixture}/corpus with spaces" \
     "FUZZ_ARTIFACT_COLLECTOR=${fixture}/collector" \
     "FUZZ_BIN_DIR=${fixture}/prebuilt" \
     "GITHUB_SHA=deadbeef" \
@@ -143,6 +167,7 @@ test_successful_targets_run_in_order() {
     "target order"
   grep -q -- '--features=unstable' "${fixture}/cargo.log" || fail "unstable feature missing"
   grep -q -- '-max_total_time=120' "${fixture}/cargo.log" || fail "fuzz time missing"
+  grep -q -- '-print_final_stats=1' "${fixture}/cargo.log" || fail "fuzz statistics missing"
   grep -q -- '-artifact_prefix=.*/targets/fuzz_create_table/libfuzzer/' "${fixture}/cargo.log" || \
     fail "target-scoped libFuzzer prefix missing"
   assert_eq success "$(jq -r '.[0].status' "${fixture}/artifacts/manifest.json")" "first status"
@@ -167,8 +192,49 @@ test_prebuilt_binary_runs_without_cargo() {
   [[ ! -e "${fixture}/cargo.log" ]] || fail "cargo ran for prebuilt binary"
   assert_eq 1 "$(wc -l <"${fixture}/binary.log" | tr -d ' ')" "prebuilt invocation count"
   grep -q -- '-max_total_time=120' "${fixture}/binary.log" || fail "prebuilt fuzz time missing"
+  grep -q -- '-print_final_stats=1' "${fixture}/binary.log" || fail "prebuilt fuzz statistics missing"
   grep -q -- '-artifact_prefix=.*/targets/fuzz_create_table/libfuzzer/' "${fixture}/binary.log" || \
     fail "prebuilt target-scoped libFuzzer prefix missing"
+  assert_eq "${fixture}/corpus with spaces/fuzz_create_table" "$(cat "${fixture}/corpus.log")" \
+    "prebuilt corpus argument"
+  assert_eq 'restored prebuilt seed' "$(cat "${fixture}/seeds.log")" "prebuilt restored input"
+  assert_file "${fixture}/corpus with spaces/fuzz_create_table/learned"
+  cleanup_fixture
+}
+
+test_cargo_reuses_separate_target_corpora() {
+  new_fixture
+  mkdir -p "${fixture}/corpus with spaces/fuzz_create_table" "${fixture}/corpus with spaces/fuzz_insert"
+  printf 'create seed\n' >"${fixture}/corpus with spaces/fuzz_create_table/seed"
+  printf 'insert seed\n' >"${fixture}/corpus with spaces/fuzz_insert/seed"
+  run_fixture $'fuzz_create_table\nfuzz_insert' false false ""
+
+  assert_eq 0 "${fixture_status}" "restored cargo run status"
+  assert_eq $'create seed\ninsert seed' "$(cat "${fixture}/seeds.log")" "target-scoped restored inputs"
+  assert_eq \
+    "fuzz_create_table"$'\t'"${fixture}/corpus with spaces/fuzz_create_table"$'\n'"fuzz_insert"$'\t'"${fixture}/corpus with spaces/fuzz_insert" \
+    "$(cat "${fixture}/corpus.log")" "cargo corpus arguments"
+  assert_eq fuzz_create_table "$(cat "${fixture}/corpus with spaces/fuzz_create_table/learned")" \
+    "first target learned input"
+  assert_eq fuzz_insert "$(cat "${fixture}/corpus with spaces/fuzz_insert/learned")" \
+    "second target learned input"
+
+  run_fixture fuzz_create_table true false ""
+  assert_eq 0 "${fixture_status}" "repeated cargo run status"
+  assert_file "${fixture}/corpus with spaces/fuzz_create_table/seed"
+  assert_file "${fixture}/corpus with spaces/fuzz_insert/learned"
+  cleanup_fixture
+}
+
+test_prebuilt_failure_preserves_corpus_and_artifacts() {
+  run_prebuilt_fixture 17
+
+  assert_eq 1 "${fixture_status}" "prebuilt failure status"
+  assert_eq 17 "$(jq -r '.[0].exit_code' "${fixture}/artifacts/manifest.json")" \
+    "prebuilt failure exit code"
+  assert_file "${fixture}/corpus with spaces/fuzz_create_table/seed"
+  assert_file "${fixture}/corpus with spaces/fuzz_create_table/learned"
+  assert_file "${fixture}/artifacts/targets/fuzz_create_table/mock-artifacts/state.txt"
   cleanup_fixture
 }
 
@@ -183,6 +249,8 @@ test_fail_fast_stops_after_first_failure() {
   assert_eq skipped_after_failure "$(jq -r '.[2].status' "${fixture}/artifacts/manifest.json")" "skipped status"
   assert_eq true "$(jq -r '.[2].after_prior_failure' "${fixture}/artifacts/manifest.json")" "skipped provenance"
   assert_file "${fixture}/artifacts/targets/fuzz_insert/mock-artifacts/state.txt"
+  assert_file "${fixture}/corpus with spaces/fuzz_insert/learned"
+  [[ ! -e "${fixture}/corpus with spaces/fuzz_alter_table" ]] || fail "skipped target corpus created"
   grep -q 'title=Fuzz target failed.*target=fuzz_insert' "${fixture}/stdout.log" || \
     fail "target failure annotation missing"
   grep -q 'title=Fuzz target skipped.*target=fuzz_alter_table' "${fixture}/stdout.log" || \
@@ -191,6 +259,8 @@ test_fail_fast_stops_after_first_failure() {
     fail "reproduction section missing"
   grep -q 'cargo +nightly fuzz run fuzz_insert' "${fixture}/artifacts/summary.md" || \
     fail "reproduction command missing"
+  grep -q 'fuzz_insert --fuzz-dir tests-fuzz' "${fixture}/artifacts/summary.md" || \
+    fail "reproduction command depends on the CI corpus path"
   cleanup_fixture
 }
 
@@ -241,6 +311,56 @@ test_invalid_configuration_fails_before_cargo() {
   cleanup_fixture
 }
 
+prepare_corpus_fixture() {
+  local targets="$1"
+  local output="$2"
+  set +e
+  FUZZ_TARGETS="${targets}" \
+    GITHUB_WORKSPACE="${fixture}/workspace with spaces" \
+    GITHUB_OUTPUT="${output}" \
+    bash "${PREPARE_CORPUS}" >"${fixture}/prepare.log" 2>&1
+  prepare_status=$?
+  set -e
+}
+
+test_corpus_cache_identity_and_paths() {
+  new_fixture
+  prepare_corpus_fixture $'fuzz_insert\r\n\nfuzz_create_table\r\nfuzz_insert\r' "${fixture}/first-output"
+  assert_eq 0 "${prepare_status}" "corpus preparation status"
+  prepare_corpus_fixture $'fuzz_create_table\nfuzz_insert' "${fixture}/second-output"
+  assert_eq 0 "${prepare_status}" "reordered corpus preparation status"
+  cmp -s "${fixture}/first-output" "${fixture}/second-output" || \
+    fail "cache identity depends on target order, duplicates, or line endings"
+  assert_eq \
+    "${fixture}/workspace with spaces/tests-fuzz/corpus/fuzz_create_table"$'\n'"${fixture}/workspace with spaces/tests-fuzz/corpus/fuzz_insert" \
+    "$(sed '1,2d; $d' "${fixture}/first-output")" "selected corpus cache paths"
+  [[ -d "${fixture}/workspace with spaces/tests-fuzz/corpus/fuzz_insert" ]] || \
+    fail "corpus directory not created"
+  [[ ! -e "${fixture}/workspace with spaces/tests-fuzz/corpus/fuzz_alter_table" ]] || \
+    fail "unselected corpus directory created"
+
+  GITHUB_SHA=another-revision prepare_corpus_fixture $'fuzz_create_table\nfuzz_insert' "${fixture}/new-revision-output"
+  cmp -s "${fixture}/first-output" "${fixture}/new-revision-output" || \
+    fail "source revision changed cache identity"
+  prepare_corpus_fixture fuzz_insert "${fixture}/subset-output"
+  assert_eq 0 "${prepare_status}" "subset corpus preparation status"
+  [[ "$(head -n 1 "${fixture}/first-output")" != "$(head -n 1 "${fixture}/subset-output")" ]] || \
+    fail "different target sets share cache identity"
+  cleanup_fixture
+}
+
+test_corpus_preparation_rejects_invalid_targets_before_writing() {
+  local targets
+  for targets in $'fuzz_insert\n../escape' 'fuzz target' $'\n\r\n'; do
+    new_fixture
+    prepare_corpus_fixture "${targets}" "${fixture}/output"
+    assert_eq 2 "${prepare_status}" "invalid corpus target status"
+    [[ ! -e "${fixture}/output" ]] || fail "invalid targets wrote cache outputs"
+    [[ ! -e "${fixture}/workspace with spaces" ]] || fail "invalid targets created directories"
+    cleanup_fixture
+  done
+}
+
 test_collector_keeps_target_scopes_separate() {
   new_fixture
   service_log="${fixture}/greptime.log"
@@ -373,10 +493,14 @@ EOF
 
 test_successful_targets_run_in_order
 test_prebuilt_binary_runs_without_cargo
+test_cargo_reuses_separate_target_corpora
+test_prebuilt_failure_preserves_corpus_and_artifacts
 test_fail_fast_stops_after_first_failure
 test_fail_fast_defaults_to_true
 test_non_fail_fast_marks_later_targets
 test_invalid_configuration_fails_before_cargo
+test_corpus_cache_identity_and_paths
+test_corpus_preparation_rejects_invalid_targets_before_writing
 test_collector_keeps_target_scopes_separate
 test_cluster_collector_honors_target_scope_and_namespace
 test_setup_failure_writes_artifact_contract

.github/scripts/run-fuzz-targets.sh

@@ -5,6 +5,7 @@ set -euo pipefail
 FUZZ_ARTIFACT_ROOT="${FUZZ_ARTIFACT_ROOT:-/tmp/greptime-fuzz-artifacts}"
 FUZZ_ARTIFACT_COLLECTOR="${FUZZ_ARTIFACT_COLLECTOR:-.github/scripts/collect-fuzz-target-artifacts.sh}"
 FUZZ_ARTIFACT_COLLECTION_TIMEOUT_SECS="${FUZZ_ARTIFACT_COLLECTION_TIMEOUT_SECS:-180}"
+FUZZ_CORPUS_ROOT="${FUZZ_CORPUS_ROOT:-tests-fuzz/corpus}"
 
 log() {
   printf '[run-fuzz-targets] %s\n' "$*"
@@ -187,19 +188,22 @@ for ((index = 0; index < ${#targets[@]}; index++)); do
   started_epoch="$(date +%s)"
   echo "::group::Fuzz target: ${target}"
   log "run target ${target}"
+  mkdir -p "${FUZZ_CORPUS_ROOT}/${target}"
+  corpus_dir="$(cd -- "${FUZZ_CORPUS_ROOT}/${target}" && pwd)"
+  log "corpus directory: ${corpus_dir}"
   if [[ -n "${FUZZ_BIN_DIR:-}" ]]; then
     fuzz_binary="${FUZZ_BIN_DIR}/${target}"
     if [[ ! -x "${fuzz_binary}" ]]; then
       log "prebuilt fuzz binary does not exist or is not executable: ${fuzz_binary}"
       exit 2
     fi
-    run_args=("${fuzz_binary}" "-max_total_time=${FUZZ_MAX_TOTAL_TIME}" "-artifact_prefix=${target_dir}/libfuzzer/")
+    run_args=("${fuzz_binary}" "${corpus_dir}" "-max_total_time=${FUZZ_MAX_TOTAL_TIME}" "-print_final_stats=1" "-artifact_prefix=${target_dir}/libfuzzer/")
   else
-    run_args=(cargo +"${RUSTUP_TOOLCHAIN:-nightly}" fuzz run "${target}" --fuzz-dir tests-fuzz -D -s none)
+    run_args=(cargo +"${RUSTUP_TOOLCHAIN:-nightly}" fuzz run "${target}" "${corpus_dir}" --fuzz-dir tests-fuzz -D -s none)
     if [[ "${fuzz_unstable}" == true ]]; then
       run_args+=(--features=unstable)
     fi
-    run_args+=(-- "-max_total_time=${FUZZ_MAX_TOTAL_TIME}" "-artifact_prefix=${target_dir}/libfuzzer/")
+    run_args+=(-- "-max_total_time=${FUZZ_MAX_TOTAL_TIME}" "-print_final_stats=1" "-artifact_prefix=${target_dir}/libfuzzer/")
   fi
   if GT_FUZZ_DUMP_DIR="${target_dir}/csv" \
     "${run_args[@]}" 2>&1 | tee "${target_dir}/fuzz.log"; then

.github/scripts/tests/ci-cache-policy.test.cjs

@@ -0,0 +1,155 @@
+const assert = require('node:assert/strict');
+const { readFileSync } = require('node:fs');
+const { resolve } = require('node:path');
+const { test } = require('node:test');
+
+const directory = resolve(__dirname, '../../workflows');
+const workflows = ['checks', 'rust', 'integration', 'nightly-ci'].map(name => ({
+  name,
+  text: readFileSync(resolve(directory, name + '.yml'), 'utf8'),
+}));
+function jobs(text) {
+  return [...text.matchAll(/^  ([\w-]+):\n([\s\S]*?)(?=^  [\w-]+:\n|(?![\s\S]))/gm)];
+}
+function steps(body) {
+  return body.split(/^      - /m).slice(1);
+}
+
+test('build caches have one controller and lightweight checks do not cache targets', () => {
+  let checked = 0;
+  for (const { name, text } of workflows) {
+    for (const [, job, body] of jobs(text)) {
+      if (!body.includes('Swatinem/rust-cache@') && !['fmt', 'check-udeps', 'toml'].includes(job)) continue;
+      for (const step of steps(body).filter(s => s.includes('uses: actions-rust-lang/setup-rust-toolchain@'))) {
+        assert.match(step, /^          cache: false$/m, name + '/' + job);
+        checked++;
+      }
+    }
+  }
+  assert.ok(checked >= 10, 'expected build and lightweight jobs to be covered');
+});
+
+test('only main writes caches, and the nightly read-only cache stays read-only', () => {
+  let checked = 0;
+  for (const { name, text } of workflows) {
+    for (const [, job, body] of jobs(text)) {
+      for (const step of steps(body)) {
+        if (!step.includes('uses: Swatinem/rust-cache@') && !step.includes('uses: actions/cache/save@')) continue;
+        const condition = step.includes('uses: Swatinem/rust-cache@')
+          ? step.match(/^          save-if: (.+)$/m)?.[1]
+          : step.match(/^        if: (.+)$/m)?.[1];
+        assert.ok(condition, name + '/' + job);
+        const expression = condition.replace(/^\$\{\{\s*|\s*\}\}$/g, '')
+          .replaceAll('steps.fuzz-binaries.outputs.cache-hit', 'hit');
+        const enabled = new Function('github', 'hit', 'return (' + expression + ');');
+        for (const [event, ref] of [
+          ['pull_request', 'refs/pull/123/merge'],
+          ['merge_group', 'refs/heads/gh-readonly-queue/main/pr-123'],
+          ['workflow_dispatch', 'refs/heads/feature'],
+        ]) assert.equal(enabled({ event_name: event, ref }, 'false'), false, name + '/' + job);
+        assert.equal(enabled({ event_name: 'schedule', ref: 'refs/heads/main' }, 'false'),
+          condition !== 'false', name + '/' + job);
+        if (step.includes('uses: actions/cache/save@')) {
+          assert.equal(enabled({ ref: 'refs/heads/main' }, 'true'), false);
+        }
+        checked++;
+      }
+    }
+  }
+  assert.ok(checked >= 10);
+});
+
+test('helper tool versions do not fragment unrelated Rust caches', () => {
+  for (const { text } of workflows) {
+    assert.doesNotMatch(text, /CARGO_(SHEAR|FUZZ)_VERSION/);
+  }
+  const integration = workflows.find(w => w.name === 'integration').text;
+  assert.match(integration, /prefix-key: "v1-fuzz-\$\{\{ env.FUZZ_VERSION \}\}"/);
+  assert.match(integration, /cargo install cargo-fuzz --version "\$\{FUZZ_VERSION\}"/);
+});
+
+// Evaluate the actual job guard as well as the cache's save condition. A valid
+// save-if alone cannot prove that a scheduled cache producer is reachable.
+function warmupJob(workflow, job, {
+  event = 'schedule', ref = 'refs/heads/main',
+  profile = event === 'workflow_dispatch' ? 'all' : undefined,
+  outputs = {}, failed = '', cancelled = false, draft = false,
+} = {}) {
+  const text = workflows.find(w => w.name === workflow).text;
+  assert.match(text, /^  schedule:$/m);
+  const body = jobs(text).find(([, name]) => name === job)?.[2];
+  assert.ok(body, workflow + '/' + job);
+  const github = { repository: 'GreptimeTeam/greptimedb', event_name: event, ref,
+    event: { pull_request: { draft } }, run_attempt: 1 };
+  const needs = { scope: { outputs }, ...Object.fromEntries(
+    ['github-script-tests', 'build', 'prepare-fuzz-targets', 'build-greptime-ci'].map(name =>
+      [name, { result: name === failed ? 'failure' : 'success' }])) };
+  const evaluate = expression => new Function('github', 'inputs', 'needs', 'cancelled',
+    'return (' + expression.replace(/^\$\{\{\s*|\s*\}\}$/g, '') + ');')(
+    github, { fuzz_profile: profile, ci_command: false }, needs, () => cancelled);
+  const admitted = evaluate(body.match(/^    if: (.+)$/m)[1]);
+  const saves = [...body.matchAll(/^          save-if: (.+)$/gm)]
+    .map(match => admitted && evaluate(match[1]));
+  return { admitted, saves };
+}
+
+const fuzzJobs = ['prepare-fuzz-targets', 'build-greptime-ci', 'fuzztest',
+  'unstable-fuzztest', 'distributed-fuzztest', 'distributed-fuzztest-with-chaos'];
+
+test('main schedules can execute cache producers and their fuzz consumers', () => {
+  for (const job of fuzzJobs) {
+    // Scheduled events do not supply workflow_dispatch inputs.
+    const result = warmupJob('integration', job, { profile: undefined });
+    assert.equal(result.admitted, true, job);
+    assert.ok(result.saves.every(Boolean), job);
+  }
+  for (const [workflow, job] of [
+    ['integration', 'build'], ['integration', 'export-import-v2-e2e'],
+    ['integration', 'prepare-fuzz-targets'], ['integration', 'build-greptime-ci'],
+    ['integration', 'distributed-fuzztest'],
+    ['rust', 'check-windows'], ['rust', 'check-riscv64'],
+    ['nightly-ci', 'sqlness-windows'], ['nightly-ci', 'test-on-windows'],
+  ]) {
+    for (const event of ['schedule', 'workflow_dispatch']) {
+      assert.deepEqual(warmupJob(workflow, job, { event }),
+        { admitted: true, saves: [true] }, workflow + '/' + job + '/' + event);
+    }
+  }
+});
+
+test('manual warming respects selected profiles without an internal command marker', () => {
+  const expected = {
+    none: [],
+    standalone: fuzzJobs.slice(0, 4),
+    distributed: [fuzzJobs[0], fuzzJobs[1], fuzzJobs[4]],
+    chaos: [fuzzJobs[0], fuzzJobs[1], fuzzJobs[5]],
+    all: fuzzJobs,
+  };
+  for (const [profile, wanted] of Object.entries(expected)) {
+    assert.deepEqual(fuzzJobs.filter(job => warmupJob('integration', job,
+      { event: 'workflow_dispatch', profile }).admitted), wanted, profile);
+  }
+  for (const job of fuzzJobs) {
+    const result = warmupJob('integration', job,
+      { event: 'workflow_dispatch', ref: 'refs/heads/feature' });
+    assert.ok(result.saves.every(value => value === false), job);
+    assert.equal(warmupJob('integration', job,
+      { event: 'pull_request', ref: 'refs/pull/1/merge', draft: true }).admitted, false, job);
+  }
+});
+
+test('warmup keeps scope exclusions, dependency failures and cancellation gates', () => {
+  const gates = ['any_fuzz', 'build_ci_needed', 'fuzz_standalone',
+    'fuzz_unstable', 'fuzz_distributed', 'fuzz_chaos'];
+  for (const [index, job] of fuzzJobs.entries()) {
+    assert.equal(warmupJob('integration', job, { failed: 'github-script-tests' }).admitted, false, job);
+    assert.equal(warmupJob('integration', job, { cancelled: true }).admitted, false, job);
+    assert.equal(warmupJob('integration', job,
+      { outputs: { [gates[index]]: 'false' } }).admitted, false, job);
+  }
+  for (const job of fuzzJobs.slice(2)) {
+    for (const failed of ['github-script-tests', 'prepare-fuzz-targets', job === 'fuzztest' ? 'build' : 'build-greptime-ci']) {
+      assert.equal(warmupJob('integration', job, { failed }).admitted, false, job);
+    }
+  }
+});

.github/scripts/tests/nightly-merge.test.cjs

@@ -0,0 +1,153 @@
+// 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 with: node --test .github/scripts/tests/nightly-merge.test.cjs
+const { readFileSync } = require('node:fs');
+const { resolve } = require('node:path');
+const assert = require('node:assert/strict');
+const { test } = require('node:test');
+const AsyncFunction = Object.getPrototypeOf(async function () {}).constructor;
+const workflow = readFileSync(resolve(__dirname, '../../workflows/nightly-merge.yml'), 'utf8');
+const scripts = [...workflow.matchAll(/          script: \|\n((?:            .*\n|\n)+)/g)]
+  .map(match => new AsyncFunction('github', 'context', 'core', 'process', match[1].replace(/^            /gm, '')));
+assert.equal(scripts.length, 2);
+const checked = '- [x] Merge during the nightly window <!-- nightly-merge -->';
+
+async function run({ sync = false, body = checked, labels = ['merge nightly'],
+  detail = {}, current = {}, threads = [[true]], count = 1, dryRun = false,
+  failFirst = false, enqueueError = Error('head changed') } = {}) {
+  const calls = [];
+  let page = 0;
+  const github = {
+    rest: {
+      pulls: { get: async () => ({ data: { state: 'open', draft: false, base: { ref: 'main' },
+        body, labels: labels.map(name => ({ name })), ...current } }) },
+      issues: {
+        listForRepo: () => {},
+        addLabels: async args => calls.push(['add', args.labels]),
+        removeLabel: async args => calls.push(['remove', args.name])
+      }
+    },
+    paginate: async () => Array.from({ length: count }, (_, i) => ({ number: i + 1, pull_request: {} })),
+    graphql: async (query, args) => {
+      if (query.includes('mutation')) {
+        calls.push(['enqueue', args]);
+        if (failFirst && calls.filter(call => call[0] === 'enqueue').length === 1) throw enqueueError;
+        return {};
+      }
+      if (query.includes('reviewThreads')) {
+        const index = page++ % threads.length;
+        calls.push(['threads', args.cursor]);
+        return { repository: { pullRequest: { reviewThreads: {
+          nodes: threads[index].map(isResolved => ({ isResolved })),
+          pageInfo: { hasNextPage: index + 1 < threads.length, endCursor: `page-${index}` }
+        } } } };
+      }
+      return { repository: { pullRequest: { id: `PR_${args.number}`, headRefOid: 'abc',
+        reviewDecision: 'APPROVED', mergeStateStatus: 'CLEAN', ...detail } } };
+    }
+  };
+  const summary = { addHeading() { return this; }, addTable(rows) { calls.push(['summary', rows]); return this; }, async write() {} };
+  await scripts[sync ? 0 : 1](github,
+    { repo: { owner: 'GreptimeTeam', repo: 'greptimedb' }, payload: { pull_request: { number: 1 } } },
+    { summary, info() {}, warning() {}, setFailed: message => calls.push(['failed', message]) },
+    { env: { DRY_RUN: String(dryRun) } });
+  return calls;
+}
+
+test('label sync grants PR label write permission without checking out PR code', () => {
+  const syncJob = workflow.split('  sync-label:\n')[1].split('  enqueue:\n')[0];
+  assert.match(syncJob, /^      pull-requests: write$/m);
+  assert.doesNotMatch(syncJob, /uses: actions\/checkout/);
+});
+
+test('template opts in and checkbox edits reconcile only the nightly label', async () => {
+  assert.ok(readFileSync(resolve(__dirname, '../../pull_request_template.md'), 'utf8').includes(checked));
+  assert.deepEqual(await run({ sync: true, labels: ['bug'] }), [['add', ['merge nightly']]]);
+  assert.deepEqual(await run({ sync: true }), []);
+  for (const body of [checked.replace('[x]', '[ ]')]) {
+    assert.deepEqual(await run({ sync: true, body }), [['remove', 'merge nightly']]);
+  }
+  assert.deepEqual(await run({ sync: true, body: checked.replace('[x]', '[X]'), labels: [] }), [['add', ['merge nightly']]]);
+});
+
+test('eligible PR is enqueued with its checked head; dry run performs no mutation', async () => {
+  assert.deepEqual((await run()).filter(c => c[0] === 'enqueue'), [['enqueue', { id: 'PR_1', sha: 'abc' }]]);
+  assert.equal((await run({ dryRun: true })).some(c => c[0] === 'enqueue'), false);
+});
+
+test('each eligibility gate independently prevents enqueueing', async () => {
+  for (const options of [
+    { labels: [] },
+    { current: { draft: true } }, { current: { state: 'closed' } }, { current: { base: { ref: 'release/v1.3' } } },
+    { detail: { reviewDecision: 'REVIEW_REQUIRED' } }, { detail: { reviewDecision: 'CHANGES_REQUESTED' } },
+    { detail: { mergeStateStatus: 'BLOCKED' } }, { detail: { mergeStateStatus: 'UNKNOWN' } },
+    { detail: { mergeQueueEntry: { id: 'entry' } } }, { detail: { autoMergeRequest: { enabledAt: 'now' } } },
+    { threads: [[false]] }, { threads: [[true], [false]] }
+  ]) {
+    assert.equal((await run(options)).some(c => c[0] === 'enqueue'), false, JSON.stringify(options));
+  }
+});
+
+test('all thread pages are checked before enqueueing', async () => {
+  const calls = await run({ threads: [[true], [true]] });
+  assert.deepEqual(calls.filter(c => c[0] === 'threads'), [['threads', null], ['threads', 'page-0']]);
+  assert.equal(calls.filter(c => c[0] === 'enqueue').length, 1);
+});
+
+test('per-run cap and individual failures do not prevent later candidates', async () => {
+  assert.equal((await run({ count: 8 })).filter(c => c[0] === 'enqueue').length, 5);
+  const calls = await run({ count: 2, failFirst: true });
+  assert.equal(calls.filter(c => c[0] === 'enqueue').length, 2);
+  assert.equal(calls.filter(c => c[0] === 'failed').length, 1);
+});
+
+test('label-only opt-in survives PRs without a template and does not require a checkbox', async () => {
+  for (const body of ['', null, 'Existing PR description', `> ${checked}`]) {
+    assert.deepEqual(await run({ sync: true, body }), []);
+    assert.deepEqual(await run({ sync: true, body, labels: [] }), []);
+    assert.equal((await run({ body })).filter(call => call[0] === 'enqueue').length, 1);
+    assert.equal((await run({ body, labels: [] })).some(call => call[0] === 'enqueue'), false);
+  }
+  // The label is authoritative for admission, even before checkbox synchronization.
+  assert.equal((await run({ body: checked.replace('[x]', '[ ]') })).some(call => call[0] === 'enqueue'), true);
+});
+
+const stackMessage = 'This pull request is part of a stack and must be enqueued using the asynchronous merge REST API. For more information, see https://docs.github.com/rest/pulls/pulls#merge-a-pull-request-asynchronously';
+
+test('stacked PR rejection is skipped without consuming the successful enqueue cap', async () => {
+  const calls = await run({ count: 7, failFirst: true,
+    enqueueError: Object.assign(Error('GraphQL request failed'), { errors: [{ message: stackMessage }] }) });
+  assert.equal(calls.some(c => c[0] === 'failed'), false);
+  assert.deepEqual(calls.find(c => c[0] === 'summary')[1], [
+    ['PR', 'Result'], ['1', 'Skipped: stacked PR'],
+    ...[2, 3, 4, 5, 6].map(number => [String(number), 'Enqueued'])
+  ]);
+  assert.deepEqual(calls.filter(c => c[0] === 'enqueue').map(c => c[1].id),
+    ['PR_1', 'PR_2', 'PR_3', 'PR_4', 'PR_5', 'PR_6']);
+});
+
+test('mixed or unstructured errors are not mistaken for a stacked PR rejection', async () => {
+  for (const enqueueError of [
+    Error(stackMessage),
+    Object.assign(Error('GraphQL request failed'), { errors: [] }),
+    Object.assign(Error('GraphQL request failed'), { errors: [{ message: stackMessage }, { message: 'Forbidden' }] })
+  ]) {
+    const calls = await run({ count: 2, failFirst: true, enqueueError });
+    assert.equal(calls.filter(c => c[0] === 'failed').length, 1);
+    assert.deepEqual(calls.find(c => c[0] === 'summary')[1], [
+      ['PR', 'Result'], ['1', 'Failed: see workflow logs'], ['2', 'Enqueued']
+    ]);
+  }
+});

.github/scripts/tests/test_metric_snapshot_s3_bench.py

@@ -28,6 +28,25 @@ SPEC.loader.exec_module(bench)
 
 
 class PhaseTests(unittest.TestCase):
+    def test_cloud_region_configuration(self):
+        for bucket, runner, accepted in (("us-east-1", "us-east-1", True),
+                                          ("ap-southeast-1", "ap-southeast-1", True),
+                                          ("us-east-1", "ap-southeast-1", False),
+                                          ("", "", False)):
+            with self.subTest(bucket=bucket, runner=runner), tempfile.TemporaryDirectory() as directory:
+                env = dict(GREPTIMEDB_IMAGE="greptime/greptimedb@sha256:" + "a" * 64,
+                           EXPECTED_COMMIT="b" * 40, RUN_PREFIX="metric-snapshot-bench/123-1/",
+                           RESULTS=str(Path(directory) / "results"), DATA_ROOT=str(Path(directory) / "data"),
+                           S3_REGION=bucket, RUNNER_REGION=runner, S3_ENDPOINT="")
+                with patch.dict(bench.os.environ, env):
+                    if accepted:
+                        benchmark = bench.Benchmark(local_smoke=False)
+                        benchmark.summary()
+                        self.assertIn(f"S3 in {bucket};", (benchmark.results / "summary.md").read_text())
+                    else:
+                        with self.assertRaisesRegex(ValueError, "same nonempty"):
+                            bench.Benchmark(local_smoke=False)
+
     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")

.github/scripts/tests/test_metric_snapshot_s3_ops.py

@@ -20,6 +20,7 @@ import io
 import json
 from pathlib import Path
 import tempfile
+import types
 import unittest
 from unittest import mock
 
@@ -82,6 +83,45 @@ class OperationsTest(unittest.TestCase):
             with self.subTest(changes=changes), self.assertRaises(ValueError):
                 ops.plan(self.env(**changes))
 
+    def test_regions_follow_configuration_and_reject_mismatch_or_empty(self):
+        for region in ("us-east-1", "ap-southeast-1"):
+            with self.subTest(region=region):
+                self.assertEqual(ops.plan(self.env(S3_REGION=region, RUNNER_REGION=region))["region"], region)
+        for bucket, runner in (("us-east-1", "ap-southeast-1"), ("", ""), (" ", " ")):
+            with self.subTest(bucket=bucket, runner=runner), self.assertRaises(ValueError):
+                ops.plan(self.env(S3_REGION=bucket, RUNNER_REGION=runner))
+
+    def test_client_uses_configured_region(self):
+        boto = types.ModuleType("boto3")
+        boto.client = mock.Mock()
+        config = types.ModuleType("botocore.config")
+        config.Config = mock.Mock()
+        with mock.patch.dict("sys.modules", {"boto3": boto, "botocore.config": config}):
+            for region in ("us-east-1", "ap-southeast-1"):
+                ops.client(region)
+                self.assertEqual(boto.client.call_args.kwargs["region_name"], region)
+            with self.assertRaises(ValueError):
+                ops.client("")
+
+    def test_hourly_estimate_controls_reservation_and_cost(self):
+        env = self.env(EC2_HOURLY_USD="2", BENCHMARK_TIMEOUT_MINUTES="30")
+        plan = ops.plan(env)
+        self.assertEqual(plan["ec2_hourly_usd"], "2")
+        self.assertEqual(plan["attempt_reserve_usd"], "5.00")
+        with mock.patch.dict(ops.os.environ, env), mock.patch.object(ops.time, "time", return_value=4600), \
+                mock.patch("sys.argv", ["ops", "cost", "--started-at", "1000"]), \
+                mock.patch("sys.stdout", new_callable=io.StringIO) as output:
+            ops.main()
+        cost = json.loads(output.getvalue())
+        self.assertEqual(cost["ec2_hourly_usd"], "2")
+        self.assertEqual(cost["ec2_estimate_usd"], "2.0000")
+        self.assertEqual(cost["campaign_estimate_with_other_reserve_usd"], "3.5000")
+        for rate in ("", "invalid", "0", "-1", "NaN", "Infinity", "-Infinity"):
+            with self.subTest(rate=rate), self.assertRaises(ValueError):
+                ops.plan(self.env(EC2_HOURLY_USD=rate))
+        with self.assertRaises(ValueError):
+            ops.plan(self.env(EC2_HOURLY_USD="4"))
+
     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")

.github/scripts/update-query-regression-helpers-image.sh

@@ -0,0 +1,14 @@
+#!/usr/bin/env bash
+set -euo pipefail
+: "${HELPERS_IMAGE:?Verified image digest is required}"
+: "${GITHUB_REPOSITORY:?Repository is required}"
+repository=greptime-registry.cn-hangzhou.cr.aliyuncs.com/tools/query-regression-helpers
+[[ "$HELPERS_IMAGE" == "$repository@sha256:"* ]]
+[[ "${HELPERS_IMAGE#"$repository@"}" =~ ^sha256:[0-9a-f]{64}$ ]]
+# Like QUERY_REGRESSION_ECS_IMAGE_ID, this is CI configuration, not a source
+# change. Create on first publication, then update only after image verification.
+gh variable set QUERY_REGRESSION_HELPERS_IMAGE \
+  --repo "$GITHUB_REPOSITORY" --body "$HELPERS_IMAGE"
+gh api "repos/$GITHUB_REPOSITORY/actions/variables/QUERY_REGRESSION_HELPERS_IMAGE" \
+  | jq -e --arg expected "$HELPERS_IMAGE" '.value == $expected' >/dev/null
+printf 'QUERY_REGRESSION_HELPERS_IMAGE updated to `%s`\n' "$HELPERS_IMAGE"