perf(rules): digest shared whole-project facts once

Whole-project facts such as project.sources repeat objects shared between
resources: every source declared in one file carries that file's contents
and entries. Digests of these facts now encode each compiler dataclass once
and refer to it by content digest, so the digest grows with distinct
content instead of with repetitions. Model and project-tree facts keep the
digest a host observes.

A real-CLI edit chain with model- and project-subject custom rules split
across hosts compares every warm compile with a cache-free oracle, and the
oracle is shown to fail when source digests are deliberately stale.

Author: kvlonge

src/sqlbuild/rule_engine/_helpers/engine/fact_replay.py

@@ -9,6 +9,7 @@
 
 import orjson
 
+from sqlbuild.rule_engine.classes.shared_fact_encoder import SharedFactEncoder
 from sqlbuild.rule_engine.classes.sql_document import SqlDocument
 from sqlbuild.rule_engine.constants import (
     FACT_AUDITS_ALL,
@@ -98,6 +99,7 @@
     FACT_TREE_READ_TEXT: 1,
 }
 _TREE_FACTS: frozenset[str] = frozenset({FACT_TREE_PATHS, *_TEXT_FACTS})
+_SHARED_FACTS: frozenset[str] = frozenset(_GLOBAL_FACTS) - _TREE_FACTS
 
 
 def fact_key_payload(key: tuple[object, ...]) -> list[str]:
@@ -123,18 +125,31 @@ def full_fact_keys(views: RuleFactViews) -> tuple[FactKey, ...]:
     return tuple(keys)
 
 
-def fact_outcome_digest(*, views: RuleFactViews, key: FactKey) -> str:
-    """Recompute one recorded fact and digest its value or raised error."""
+def fact_outcome_digest(
+    *, views: RuleFactViews, key: FactKey, shared: SharedFactEncoder | None = None
+) -> str:
+    """Recompute one recorded fact and digest its value, sharing whole-project encodings."""
 
     try:
         value: object = _fact_value(views=views, key=key)
     except FactDigestError:
         raise
     except Exception as error:
         return fact_error_digest(error)
+    if shared is not None and key[0] in _SHARED_FACTS:
+        try:
+            return hashlib.sha256(shared.encode(("value", value))).hexdigest()
+        except (TypeError, FactDigestError) as error:
+            raise FactDigestError(str(error)) from error
     return fact_value_digest(value)
 
 
+def shared_fact_encoder() -> SharedFactEncoder:
+    """Return an encoder for whole-project facts that share compiler objects."""
+
+    return SharedFactEncoder(options=_ORJSON_OPTIONS, encode_default=_encode_default)
+
+
 def fact_value_digest(value: object) -> str:
     """Digest one fact value exactly as a recomputed fact would digest it."""
 

src/sqlbuild/rule_engine/classes/fact_digests.py

@@ -8,7 +8,11 @@
 
 import orjson
 
-from sqlbuild.rule_engine._helpers.engine.fact_replay import fact_outcome_digest
+from sqlbuild.rule_engine._helpers.engine.fact_replay import (
+    fact_outcome_digest,
+    shared_fact_encoder,
+)
+from sqlbuild.rule_engine.classes.shared_fact_encoder import SharedFactEncoder
 from sqlbuild.rule_engine.models import RuleFactViews
 from sqlbuild.rule_engine.types import FactKey
 
@@ -20,6 +24,7 @@ def __init__(self, *, views: Callable[[], RuleFactViews]) -> None:
         self._views_factory: Callable[[], RuleFactViews] = views
         self._views: RuleFactViews | None = None
         self._digests: dict[FactKey, str] = {}
+        self._shared: SharedFactEncoder = shared_fact_encoder()
 
     @property
     def views(self) -> RuleFactViews:
@@ -33,7 +38,7 @@ def digest(self, key: FactKey) -> str:
         cached: str | None = self._digests.get(key)
         if cached is not None:
             return cached
-        digest: str = fact_outcome_digest(views=self.views, key=key)
+        digest: str = fact_outcome_digest(views=self.views, key=key, shared=self._shared)
         self._digests[key] = digest
         return digest
 

src/sqlbuild/rule_engine/classes/shared_fact_encoder.py

@@ -0,0 +1,41 @@
+"""Digest whole-project fact values whose resources share large compiler objects."""
+
+from __future__ import annotations
+
+import dataclasses
+import hashlib
+from collections.abc import Callable
+
+import orjson
+
+from sqlbuild.rule_engine.constants import SHARED_FACT_DATACLASS_TAG
+
+
+class SharedFactEncoder:
+    """Encode each compiler dataclass once and refer to it by digest wherever it repeats."""
+
+    def __init__(self, *, options: int, encode_default: Callable[[object], object]) -> None:
+        self._options: int = options | orjson.OPT_PASSTHROUGH_DATACLASS
+        self._encode_default: Callable[[object], object] = encode_default
+        self._fields: dict[type, tuple[str, ...]] = {}
+        self._digests: dict[int, tuple[object, str]] = {}
+
+    def encode(self, value: object) -> bytes:
+        return orjson.dumps(value, default=self._default, option=self._options)
+
+    def _default(self, value: object) -> object:
+        kind: type = type(value)
+        names: tuple[str, ...] | None = self._fields.get(kind)
+        if names is None:
+            if not dataclasses.is_dataclass(kind):
+                return self._encode_default(value)
+            names = tuple(field.name for field in dataclasses.fields(kind))
+            self._fields[kind] = names
+        cached: tuple[object, str] | None = self._digests.get(id(value))
+        if cached is None:
+            encoded: bytes = self.encode(
+                [f"{kind.__module__}.{kind.__qualname__}", [getattr(value, name) for name in names]]
+            )
+            cached = (value, hashlib.sha256(encoded).hexdigest())
+            self._digests[id(value)] = cached
+        return [SHARED_FACT_DATACLASS_TAG, cached[1]]

src/sqlbuild/rule_engine/constants.py

@@ -16,6 +16,7 @@
 NATIVE_RULES_MEMO_VERSION: str = "native-rules-response-v2"
 NATIVE_RULES_MEMO_FILE: str = "target/rules-cache/bulk/native-response.json"
 CUSTOM_HOST_INPUT_TUPLE_SIZE: int = 2
+SHARED_FACT_DATACLASS_TAG: str = "\0dataclass"
 CUSTOM_HOST_ABANDONED_SUFFIX: str = ".abandoned"
 CUSTOM_HOST_INPUT_POLL_SECONDS: float = 0.002
 CUSTOM_HOST_INPUT_WAIT_SECONDS: float = 120.0

tests/e2e/src/sqlbuild/cli/commands/main/rules/_test_types.py

@@ -222,3 +222,13 @@ class RulesEditChainCase:
     edits: tuple[RulesCacheEditCase, ...]
     expected_cold_codes: tuple[str, ...]
     expected_exit_code: int = 1
+
+
+@dataclass(frozen=True)
+class BrokenInvalidationChainCase:
+    """An edit chain compiled with deliberately stale fact digests against a cache-free oracle."""
+
+    description: str
+    edits: tuple[RulesCacheEditCase, ...]
+    stale_digests_sitecustomize: str
+    expected_first_divergent_edit: str

tests/e2e/src/sqlbuild/cli/commands/main/rules/helpers.py

@@ -3,6 +3,7 @@
 import hashlib
 import json
 import os
+import shutil
 import subprocess
 import sys
 from collections.abc import Callable
@@ -411,6 +412,121 @@ def flagged_values(*, model: Model, ctx: RuleContext) -> list[Finding]:
 }
 
 
+_ORDER_SLICE_COUNT: int = 45
+CUSTOM_RULES_EDIT_PROJECT: dict[str, str] = {
+    "sqlbuild_project.toml": (
+        'name = "orders"\nadapter = "duckdb"\n\n'
+        '[connection]\ndatabase = "warehouse.duckdb"\n\n'
+        '[rules]\nselect = ["XSQBR"]\n\n'
+        "[rules.thresholds]\nmin_custom_rule_test_cases = 0\n"
+    ),
+    "sources/raw.yml": (
+        "sources:\n"
+        "  - name: raw_orders\n    description: Raw orders.\n"
+        '    expression: "(SELECT 1 AS id, CAST(10.5 AS DOUBLE) AS amount, 1 AS customer_id)"\n'
+        "  - name: raw_customers\n    description: Raw customers.\n"
+        '    expression: "(SELECT 1 AS customer_id)"\n'
+        "  - name: raw_products\n    description: Raw products.\n"
+        '    expression: "(SELECT 1 AS product_id)"\n'
+    ),
+    "models/staging/stg_orders.sql": (
+        'MODEL (description "Staged orders");\n\n'
+        'SELECT id AS order_id, amount, customer_id FROM __source("raw_orders")\n'
+    ),
+    "models/staging/stg_customers.sql": (
+        'MODEL (description "Staged customers");\n\n'
+        'SELECT customer_id FROM __source("raw_customers")\n'
+    ),
+    "models/marts/order_totals.sql": (
+        'MODEL (description "Order totals");\n\n'
+        'SELECT order_id, amount * 2 AS doubled_amount FROM __ref("stg_orders") WHERE amount > 7\n'
+    ),
+    **{
+        f"models/intermediate/orders_{index:03d}.sql": (
+            f'MODEL (description "Order slice {index}");\n\n'
+            'SELECT order_id FROM __ref("stg_orders")\n'
+        )
+        for index in range(_ORDER_SLICE_COUNT)
+    },
+    "rules/approved_sources.yaml": "raw_orders\nraw_customers\n",
+    "rules/orders.py": """from sqlbuild.rules import Finding, Model, Project, RuleContext, rule
+
+
+@rule(code="XSQBRORD101", message="Filters hard-code an amount", remediation="Use a constant.")
+def high_amount_filters(*, model: Model, ctx: RuleContext) -> list[Finding]:
+    expanded: str = ctx.sql.for_model(model).expanded.source
+    return [ctx.finding(subject=model)] if "amount > 9" in expanded else []
+
+
+@rule(code="XSQBRORD102", message="Staging models need a consumer", remediation="Use it.")
+def staging_consumers(*, model: Model, ctx: RuleContext) -> list[Finding]:
+    if model.path.parent.name != "staging":
+        return []
+    return [] if ctx.graph.dependents(model) else [ctx.finding(subject=model)]
+
+
+@rule(code="XSQBRORD103", message="Marts must enforce a contract", remediation="Enforce it.")
+def mart_contracts(*, model: Model, ctx: RuleContext) -> list[Finding]:
+    if model.path.parent.name != "marts":
+        return []
+    return [] if ctx.contracts.enforced(model) else [ctx.finding(subject=model)]
+
+
+@rule(code="XSQBRORD201", message="Sources must be approved", remediation="Approve it.")
+def approved_sources(*, project: Project, ctx: RuleContext) -> list[Finding]:
+    del project
+    approved: set[str] = set(ctx.project.tree.read_text("rules/approved_sources.yaml").split())
+    names: list[str] = sorted(str(getattr(source, "name")) for source in ctx.project.sources)
+    return [
+        ctx.finding(subject="sources/raw.yml", message=f"Source {name} is not approved")
+        for name in names
+        if name not in approved
+    ]
+
+
+@rule(code="XSQBRORD202", message="Marts must stay few", remediation="Consolidate them.")
+def few_marts(*, project: Project, ctx: RuleContext) -> list[Finding]:
+    del project
+    marts: list[Model] = [item for item in ctx.project.models if item.path.parent.name == "marts"]
+    if len(marts) < 2:
+        return []
+    return [ctx.finding(subject=marts[0], message=f"{len(marts)} marts")]
+""",
+}
+STALE_SOURCE_FACTS_SITECUSTOMIZE: str = """from sqlbuild.rule_engine.classes import fact_digests
+
+_digest = fact_digests.FactDigests.digest
+
+
+def _stale_sources(self, key):
+    return "stale" if key == ("project.sources",) else _digest(self, key)
+
+
+fact_digests.FactDigests.digest = _stale_sources
+"""
+
+
+def custom_rules_edit_chain(
+    *, root: Path, edits: tuple[Any, ...], environment: tuple[tuple[str, str], ...] = ()
+) -> tuple[
+    tuple[int, str, str], tuple[tuple[int, str, str], ...], tuple[tuple[int, str, str], ...]
+]:
+    """Compile the custom-rule project cold, then after each edit beside a cache-free oracle."""
+
+    project_dir: Path = root / "orders"
+    write_project_files(project_dir=project_dir, files=CUSTOM_RULES_EDIT_PROJECT)
+    cold: tuple[int, str, str] = rules_compile_outcome(project_dir, environment=environment)
+    edited: list[tuple[int, str, str]] = []
+    oracles: list[tuple[int, str, str]] = []
+    for step, edit_case in enumerate(edits):
+        edit_case.edit(project_dir)
+        edited.append(rules_compile_outcome(project_dir, environment=environment))
+        oracle_dir: Path = root / f"oracle_{step}"
+        shutil.copytree(project_dir, oracle_dir, ignore=shutil.ignore_patterns("target"))
+        oracles.append(rules_compile_outcome(oracle_dir, "--no-cache"))
+    return cold, tuple(edited), tuple(oracles)
+
+
 def write_project_files(*, project_dir: Path, files: dict[str, str]) -> None:
     """Write authored project files beneath one project directory."""
 
@@ -420,7 +536,9 @@ def write_project_files(*, project_dir: Path, files: dict[str, str]) -> None:
         path.write_text(contents, encoding="utf-8")
 
 
-def rules_compile_outcome(project_dir: Path, *arguments: str) -> tuple[int, str, str]:
+def rules_compile_outcome(
+    project_dir: Path, *arguments: str, environment: tuple[tuple[str, str], ...] = ()
+) -> tuple[int, str, str]:
     """Compile in a fresh process and return its exit code, diagnostics, and artifacts digest."""
 
     result: subprocess.CompletedProcess[str] = subprocess.run(
@@ -436,6 +554,7 @@ def rules_compile_outcome(project_dir: Path, *arguments: str) -> tuple[int, str,
         capture_output=True,
         text=True,
         check=False,
+        env={**os.environ, **dict(environment)},
     )
     payload: dict[str, Any] = json.loads(result.stdout or "{}")
     artifacts: Any = hashlib.sha256()