test: guard varied warm compilation and edits

.github/workflows/verify.yml

@@ -419,6 +419,26 @@ jobs:
             make test-e2e-cache-compile-performance \
               SQLBUILD_BENCHMARK_MODELS="${BENCHMARK_MODELS}"
 
+      - name: Run varied 1k warm-cache guard under 4 GiB
+        if: matrix.models == 3000
+        env:
+          MEMORY_LIMIT_BYTES: "4294967296"
+        run: |
+          sudo systemd-run \
+            --unit="sqlbuild-varied-cache-${GITHUB_RUN_ID}" \
+            --uid="$(id -u)" \
+            --gid="$(id -g)" \
+            --same-dir \
+            --setenv="HOME=${HOME}" \
+            --setenv="PATH=${PATH}" \
+            --setenv="SQLBUILD_REQUIRE_MEMORY_LIMIT_BYTES=${MEMORY_LIMIT_BYTES}" \
+            --property="MemoryMax=${MEMORY_LIMIT_BYTES}" \
+            --property="MemorySwapMax=0" \
+            --wait \
+            --collect \
+            --pipe \
+            make test-e2e-varied-cache-performance
+
   rules_performance:
     name: Rules performance guards
     needs: [changes, lockfile]

Makefile

@@ -6,7 +6,7 @@ SHELL := /bin/bash
 	test-e2e-duckdb-cli-data test-e2e-duckdb-cli test-e2e-duckdb-virtual \
 	test-e2e-duckdb-integrations test-e2e-performance \
 	test-e2e-cold-compile-performance test-e2e-cache-compile-performance \
-	test-e2e-dense-compile-performance
+	test-e2e-dense-compile-performance test-e2e-varied-cache-performance
 
 format:
 	uv run ruff format .
@@ -255,6 +255,12 @@ test-e2e-cache-compile-performance:
 		-vv --log-cli-level=INFO --color=yes
 
 
+test-e2e-varied-cache-performance:
+	env PYTHONUNBUFFERED=1 SQLBUILD_CONCURRENCY=4 uv run pytest \
+		tests/e2e/src/sqlbuild/cli/commands/main/compile/test_varied_compile_cache_performance.py \
+		-n auto --dist loadfile -m cache_compile_performance \
+		-vv -rP --log-level=INFO --log-cli-level=INFO --color=yes
+
 test-virtual:
 	@mkdir -p /tmp/opencode
 	@log="/tmp/opencode/test-virtual-$$(date +%Y%m%d-%H%M%S).log"; \

docs/dense-compile-performance.md

@@ -74,6 +74,10 @@ Custom rules continue to run in their isolated host. Disabled rules caches avoid
 unused fact fingerprints; enabled caches retain their normal identities and
 invalidation behavior.
 
+The [varied warm-compile guard](warm-compile-performance.md) adds operator-shape
+diversity and a shared dependency graph, measuring unchanged, leaf, upstream and
+macro-edit compiles against uncached semantic/artifact oracles.
+
 This fixture protects dense query analysis, declaration expansion, rules and
 artifact generation. It is not an exact reproduction of every application:
 dynamic plugins, incremental materialization mixes and deeply nested query

docs/warm-compile-performance.md

@@ -0,0 +1,48 @@
+# Varied warm-compile guard
+
+`make test-e2e-varied-cache-performance` exercises a generated 1,000-model order
+project in fresh CLI processes. The cache performance CI job runs this guard in a
+4GiB cgroup with swap disabled, alongside the existing scaling benchmarks.
+
+The workload uses independently authored synthetic SQL. It retains the dense
+fixture's sources, seeds, SQL functions, audits, scoped Python macros, SQL tests,
+all built-in rules and one isolated custom project rule. It adds a shared dependency
+graph and varies the actual expressions: CASE, aggregate windows, array indexing,
+null handling, grouped queries, lateral joins and unions.
+
+## Structural protection
+
+The guard measures structure from compiled SQL and model dependencies. It requires:
+
+- At least 950 distinct operator-tree shapes. Identifier names and literal values
+  do not count toward this diversity measurement.
+- A connected component containing at least 900 models.
+- Dependency depth between 40 and 60 models, with 150–400 leaf models.
+
+These checks prevent a fixture change from silently replacing the workload with
+mostly repeated queries or independent shallow chains. Structural diversity is
+still a proxy for application workloads; private acceptance measurements remain
+separate from this public synthetic fixture.
+
+## Cache and correctness protection
+
+The guard measures initial cache fill, an unchanged warm compile, a leaf SQL edit,
+a shared upstream SQL edit and a Python macro edit. Each edit is followed by an
+unchanged compile and an uncached oracle with both analysis and rule cache reuse
+disabled. Project configuration is restored after each oracle.
+
+Every edited cached result must match its oracle's semantic JSON and every compiled
+artifact. Unchanged runs must reuse analysis for every model. Edit runs must record
+both reuse and invalidation, rather than treating a high cache-hit count alone as
+evidence of correctness. A smaller integration fixture executes its generated SQL
+tests in DuckDB with the standard rules enabled.
+
+Warm and edit budgets apply to complete fresh-process CLI wall time, including
+discovery, macro expansion, lint, rules, artifacts and output. SQL-analysis timing
+alone is insufficient: these other phases can dominate after analysis is cached.
+
+The initial ceilings are 24s for cache fill, 10s for unchanged warm compiles and
+13s for edits, with a 2GiB process-RSS ceiling and the separate 4GiB combined CI
+limit. These are regression ceilings with hosted-execution headroom, not promised
+application timings. Golden fingerprints also protect the baseline and edited
+outputs across compiler releases.

scripts/cold_compile_performance/_helpers/varied_benchmark.py

@@ -0,0 +1,117 @@
+"""Measure diverse cached compiler workloads against uncached semantic oracles."""
+
+from itertools import filterfalse
+from operator import attrgetter
+from pathlib import Path
+
+from scripts.cold_compile_performance._helpers.dense_project import dense_model_name
+from scripts.cold_compile_performance._helpers.varied_project import write_varied_compile_project
+from scripts.cold_compile_performance.exceptions import CompileBenchmarkFixtureError
+from sqlbuild.rule_engine._helpers.engine.catalogue import build_catalogue, select_rules
+from sqlbuild.rule_engine.main.load_config import load_rules_config
+from sqlbuild.rule_engine.models import Rule, RulesConfig
+from tests.e2e.src.sqlbuild.cli.commands.main.compile._test_types import VariedCompileCacheTestCase
+from tests.e2e.src.sqlbuild.cli.commands.main.compile.helpers import (
+    FreshProcessCompileBenchmarkResult,
+    _run_fresh_process_compile_benchmark,
+)
+
+
+def run_varied_cache_benchmark(
+    *, project_dir: Path, test_case: VariedCompileCacheTestCase
+) -> dict[str, FreshProcessCompileBenchmarkResult]:
+    write_varied_compile_project(project_dir=project_dir, model_count=test_case.model_count)
+    _assert_complete_rules(project_dir)
+    measurements: dict[str, FreshProcessCompileBenchmarkResult] = {}
+    measurements["cold"] = _measure(
+        project_dir=project_dir, label="cold", budget=test_case.expected_cold_max_seconds
+    )
+    measurements["warm"] = _measure(
+        project_dir=project_dir, label="warm", budget=test_case.expected_warm_max_seconds
+    )
+    leaf: Path = next(
+        (project_dir / "models").rglob(f"{dense_model_name(test_case.model_count - 1)}.sql")
+    )
+    _edit_model(leaf)
+    measurements.update(_measure_edit(project_dir=project_dir, label="leaf", test_case=test_case))
+    shared: Path = next((project_dir / "models").rglob(f"{dense_model_name(0)}.sql"))
+    _edit_model(shared)
+    measurements.update(_measure_edit(project_dir=project_dir, label="shared", test_case=test_case))
+    macro: Path = next((project_dir / "models").rglob("offset00000.py"))
+    contents: str = macro.read_text(encoding="utf-8")
+    if contents.count('return f"({expression} + 1)"') != 1:
+        raise CompileBenchmarkFixtureError("Expected one editable macro expression")
+    macro.write_text(
+        contents.replace('return f"({expression} + 1)"', 'return f"({expression} + 2)"'),
+        encoding="utf-8",
+    )
+    measurements.update(_measure_edit(project_dir=project_dir, label="macro", test_case=test_case))
+    return measurements
+
+
+def _assert_complete_rules(project_dir: Path) -> None:
+    config: RulesConfig = load_rules_config(project_dir=project_dir)
+    catalogue: tuple[Rule, ...] = build_catalogue(config=config, project_dir=project_dir)
+    selected: tuple[Rule, ...] = select_rules(
+        catalogue=catalogue, config=config, project_dir=project_dir
+    )
+    builtin_codes: set[str] = {rule.code for rule in filterfalse(attrgetter("custom"), catalogue)}
+    selected_codes: set[str] = {rule.code for rule in filterfalse(attrgetter("custom"), selected)}
+    if (
+        selected_codes != builtin_codes
+        or sum(rule.custom for rule in selected) != 1
+        or config.ignore
+        or config.rule_exceptions
+        or config.rule_ignores
+        or config.thresholds
+    ):
+        raise CompileBenchmarkFixtureError(
+            "Expected all built-in rules and one unrelaxed custom rule"
+        )
+
+
+def _edit_model(path: Path) -> None:
+    contents: str = path.read_text(encoding="utf-8")
+    if contents.count("CAST(id AS INTEGER)") != 1:
+        raise CompileBenchmarkFixtureError("Expected one editable model projection")
+    path.write_text(
+        contents.replace("CAST(id AS INTEGER)", "CAST(COALESCE(id, id) AS INTEGER)"),
+        encoding="utf-8",
+    )
+
+
+def _measure_edit(
+    *,
+    project_dir: Path,
+    label: str,
+    test_case: VariedCompileCacheTestCase,
+) -> dict[str, FreshProcessCompileBenchmarkResult]:
+    measurements: dict[str, FreshProcessCompileBenchmarkResult] = {}
+    measurements[label] = _measure(
+        project_dir=project_dir, label=label, budget=test_case.expected_edit_max_seconds
+    )
+    measurements[f"after_{label}"] = _measure(
+        project_dir=project_dir, label=f"after_{label}", budget=test_case.expected_warm_max_seconds
+    )
+    config: Path = project_dir / "sqlbuild_project.toml"
+    original: bytes = config.read_bytes()
+    try:
+        config.write_bytes(original + b"\n[rules.cache]\nenabled = false\n")
+        measurements[f"oracle_{label}"] = _run_fresh_process_compile_benchmark(
+            project_dir=project_dir,
+            label=f"oracle-{label}",
+            expected_max_wall_seconds=test_case.expected_cold_max_seconds,
+            compile_args=("--no-cache",),
+        )
+    finally:
+        config.write_bytes(original)
+    return measurements
+
+
+def _measure(*, project_dir: Path, label: str, budget: float) -> FreshProcessCompileBenchmarkResult:
+    return _run_fresh_process_compile_benchmark(
+        project_dir=project_dir,
+        label=f"varied-{label}",
+        expected_max_wall_seconds=budget,
+        compile_args=(),
+    )

scripts/cold_compile_performance/_helpers/varied_project.py

@@ -0,0 +1,138 @@
+"""Generate diverse order-query shapes over a shared, bounded dependency graph."""
+
+import re
+from pathlib import Path
+
+from scripts.cold_compile_performance._helpers.dense_project import (
+    dense_model_name,
+    write_dense_compile_project,
+)
+from scripts.cold_compile_performance.constants import (
+    VARIED_ARRAY_VARIANT,
+    VARIED_CASE_SHARE,
+    VARIED_GROUP_SHARE,
+    VARIED_WINDOW_SHARE,
+    VARIED_WINDOW_VARIANTS,
+)
+
+_QUANTITY_EXPRESSION: str = "COALESCE(b.amount, 0) + CAST(b.id AS DOUBLE)"
+_QUANTITY_VARIANTS: tuple[str, ...] = (
+    _QUANTITY_EXPRESSION,
+    "CASE WHEN b.id IS NOT NULL THEN b.amount + b.id END",
+    "ABS(b.amount) + ABS(b.id)",
+    "SUM(b.amount) OVER (PARTITION BY b.id) + b.id",
+    "MAX(b.amount) OVER (PARTITION BY b.id) + b.id",
+    "COALESCE(NULLIF(b.amount, 0), b.id) + b.id",
+    "[b.amount, b.amount][1] + b.id",
+    "GREATEST(b.amount, b.id) + LEAST(b.amount, b.id)",
+)
+_IMPORT: re.Pattern[str] = re.compile(r'__(?:ref|source)\("([^"\n]+)"\)')
+
+
+def write_varied_compile_project(*, project_dir: Path, model_count: int) -> None:
+    """Write independent neutral SQL with diverse expressions and shared ancestors."""
+
+    write_dense_compile_project(project_dir=project_dir, model_count=model_count)
+    for index, model in enumerate(sorted((project_dir / "models").rglob("*.sql"))):
+        contents: str = model.read_text(encoding="utf-8")
+        replacements: dict[str, str] = _dependency_replacements(
+            index=index, contents=contents, model_count=model_count
+        )
+        for previous, current in replacements.items():
+            contents = contents.replace(previous, current)
+        contents = _vary_expressions(index=index, contents=contents)
+        contents = _vary_query_structure(index=index, contents=contents)
+        model.write_text(contents, encoding="utf-8")
+        test_name: str = (
+            f"test_{dense_model_name(index).replace('__', '_')}__preserves_order_id.sql"
+        )
+        test: Path = next((project_dir / "tests").rglob(test_name))
+        test_sql: str = test.read_text(encoding="utf-8")
+        for previous, current in replacements.items():
+            test_sql = test_sql.replace(_mock_name(previous), _mock_name(current))
+        test.write_text(test_sql, encoding="utf-8")
+
+
+def _vary_expressions(*, index: int, contents: str) -> str:
+    fragments: list[str] = contents.split(_QUANTITY_EXPRESSION)
+    output: list[str] = [fragments[0]]
+    for position, fragment in enumerate(fragments[1:]):
+        selector: int = ((index + 1) * 2654435761) ^ ((position + 1) * 2246822519)
+        variant: int = (selector >> ((position + 1) % 19)) % len(_QUANTITY_VARIANTS)
+        if variant == VARIED_ARRAY_VARIANT and index % 5 != 0:
+            variant = 0
+        if variant in VARIED_WINDOW_VARIANTS and index % 20 >= VARIED_WINDOW_SHARE:
+            variant = 2
+        if variant == 1 and index % 10 >= VARIED_CASE_SHARE:
+            variant = 5
+        output.extend((_QUANTITY_VARIANTS[variant], fragment))
+    return "".join(output)
+
+
+def _vary_query_structure(*, index: int, contents: str) -> str:
+    if index % 4 != 1:
+        contents = re.sub(
+            r"combined AS \(\n(SELECT [^\n]+ FROM projected)\n.*?\n\)",
+            r"combined AS (\n\1\n)",
+            contents,
+            flags=re.DOTALL,
+        )
+    prefix, marker, tail = contents.partition("projected AS (")
+    parts: tuple[str, str, str] = tail.partition("),\ncombined AS (")
+    projected: str = parts[0]
+    separator: str = parts[1]
+    rest: str = parts[2]
+    if index % 7 == 0:
+        projected = projected.replace("b.amount", "quantities.quantity", 1)
+        projected += (
+            "\nINNER JOIN LATERAL (SELECT b.amount AS quantity) AS quantities"
+            " ON quantities.quantity = b.amount"
+        )
+    if index % 10 < VARIED_GROUP_SHARE:
+        keys: list[str] = ["b.id", "b.amount"]
+        if index % 7 == 0:
+            keys.append("quantities.quantity")
+        keys.extend(re.findall(r"COALESCE\((l\d+\.amount), 0\)", projected))
+        projected += "\nGROUP BY " + ", ".join(dict.fromkeys(keys))
+    return prefix + marker + projected + separator + rest
+
+
+def _dependency_replacements(*, index: int, contents: str, model_count: int) -> dict[str, str]:
+    cohort_start: int = index // 1000 * 1000
+    local_index: int = index - cohort_start
+    cohort_size: int = min(1000, model_count - cohort_start)
+    roots: int = max(1, cohort_size * 14 // 100)
+    width: int = max(1, cohort_size * 17 // 1000)
+    source_count: int = max(1, cohort_size * 24 // 100)
+    imports: list[str] = list(dict.fromkeys(match.group(0) for match in _IMPORT.finditer(contents)))
+    replacements: dict[str, str] = {}
+    used: set[int] = set()
+    for position, imported in enumerate(imports):
+        if local_index < roots:
+            replacements[imported] = (
+                f'__source("orders_{cohort_start * 24 // 100 + local_index % source_count:05d}")'
+                if position == 0
+                else imported
+            )
+        elif position == 0:
+            level: int = (local_index - roots) // width
+            start: int = roots + (level - 1) * width if level else 0
+            count: int = width if level else roots
+            slot: int = local_index - local_index % 3 if local_index % 3 == 1 else local_index
+            parent: int = cohort_start + start + (slot * 7 + 3) % count
+            used.add(parent)
+            replacements[imported] = f'__ref("{dense_model_name(parent)}")'
+        elif position % 2 == 0:
+            parent = cohort_start + (local_index * 31 + position * 13) % roots
+            if parent not in used:
+                used.add(parent)
+                replacements[imported] = f'__ref("{dense_model_name(parent)}")'
+    return replacements
+
+
+def _mock_name(reference: str) -> str:
+    return (
+        reference.replace('__ref("', "__ref__")
+        .replace('__source("', "__source__")
+        .removesuffix('")')
+    )