fix(native): compare catalog shapes on cold compiles and record deferrals after the session

Keep the binding catalog's schemas and analysis shapes in captures and
ignore them only on warm stages, where Python's analysis cache skips
catalog preparation. Record per-model deferrals once the session
completes, and fall back to Python when provide or finish panics.

Author: kvlonge

crates/sqlbuild-python/src/bindings/_helpers/analysis_session/session.rs

@@ -15,6 +15,8 @@ use sqlbuild_analysis::assembly::analysis_session::models::{
 use sqlbuild_analysis::assembly::analysis_session::types::{Pairs, Shapes};
 use sqlbuild_analysis::semantic_validation::types::DiagnosticRow;
 
+use sqlbuild_core::panics::main::catch_compiler_panic::catch_compiler_panic;
+
 use crate::bindings::_helpers::boundary::panics::compiler_error;
 use crate::bindings::models::ProjectCatalog;
 use crate::bindings::types::CompilerDetach;
@@ -124,16 +126,18 @@ impl NativeModelAnalysisSession {
             return false;
         };
         let answers: Vec<DeferredAnalysis> = results.into_iter().map(deferred_analysis).collect();
-        let provided: Result<(), String> = provide_deferred_analyses(&mut session, answers);
+        let provided: Result<(), String> =
+            catch_compiler_panic(|| provide_deferred_analyses(&mut session, answers));
         self.inner = Some(session);
         self.kept(provided).is_some()
     }
 
     /// Every model's outcome once the session is done, or None to analyse in Python.
     fn finish(&mut self) -> Option<FinishRow> {
         let session: AnalysisSession = self.inner.take()?;
-        let outcome: Result<SessionOutcome, String> = finish_analysis_session(session);
-        self.kept(outcome).map(finish_row)
+        let outcome: Result<FinishRow, String> =
+            catch_compiler_panic(|| finish_analysis_session(session).map(finish_row));
+        self.kept(outcome)
     }
 
     /// Why the session handed the analysis back to Python, when it did.

scripts/compiler_differential/_helpers/comparing/capture.py

@@ -17,8 +17,10 @@
 from scripts.compiler_differential.models import Divergence
 
 
-def first_capture_difference(*, left: Path, right: Path) -> Divergence | None:
-    """Return the first logical JSON pointer, or line for invalid JSON, where captures differ."""
+def first_capture_difference(
+    *, left: Path, right: Path, ignored: frozenset[str] = frozenset()
+) -> Divergence | None:
+    """Return the first JSON pointer outside `ignored`, or line for invalid JSON, that differs."""
 
     try:
         left_capture: CaptureFile = CaptureFile(left)
@@ -27,6 +29,7 @@ def first_capture_difference(*, left: Path, right: Path) -> Divergence | None:
             left=left_capture.root,
             right=right_capture.root,
             resolve=(left_capture.resolve, right_capture.resolve),
+            ignored=ignored,
         )
     except (json.JSONDecodeError, UnicodeDecodeError):
         return _first_line_difference(left=left, right=right)

scripts/compiler_differential/_helpers/comparing/compare.py

@@ -13,9 +13,14 @@
 
 
 def first_json_difference(
-    *, left: object, right: object, pointer: str = "", resolve: NodeResolvers | None = None
+    *,
+    left: object,
+    right: object,
+    pointer: str = "",
+    resolve: NodeResolvers | None = None,
+    ignored: frozenset[str] = frozenset(),
 ) -> Divergence | None:
-    """Return the first JSON pointer whose value, type, or key order differs, via `resolve`."""
+    """Return the first JSON pointer outside `ignored` whose value, type or key order differs."""
 
     if resolve is not None:
         left_digest: str | None = shared_digest(left)
@@ -28,12 +33,16 @@ def first_json_difference(
     right_object: dict[str, object] | None = as_json_object(right)
     if left_object is not None and right_object is not None:
         return _first_object_difference(
-            left=left_object, right=right_object, pointer=pointer, resolve=resolve
+            left=left_object, right=right_object, pointer=pointer, resolve=resolve, ignored=ignored
         )
     if isinstance(left, list) and isinstance(right, list):
         for index, (left_item, right_item) in enumerate(zip(left, right, strict=False)):
             found: Divergence | None = first_json_difference(
-                left=left_item, right=right_item, pointer=f"{pointer}/{index}", resolve=resolve
+                left=left_item,
+                right=right_item,
+                pointer=f"{pointer}/{index}",
+                resolve=resolve,
+                ignored=ignored,
             )
             if found is not None:
                 return found
@@ -99,17 +108,20 @@ def _first_object_difference(
     right: dict[str, object],
     pointer: str,
     resolve: NodeResolvers | None,
+    ignored: frozenset[str],
 ) -> Divergence | None:
     for key in (*left, *(key for key in right if key not in left)):
         location: str = f"{pointer}/{json_pointer_token(key)}"
+        if location in ignored:
+            continue
         if key not in left or key not in right:
             return _divergence(
                 location=location,
                 left=_shown(value=left.get(key, MISSING_VALUE), resolve=resolve, side=0),
                 right=_shown(value=right.get(key, MISSING_VALUE), resolve=resolve, side=1),
             )
         found: Divergence | None = first_json_difference(
-            left=left[key], right=right[key], pointer=location, resolve=resolve
+            left=left[key], right=right[key], pointer=location, resolve=resolve, ignored=ignored
         )
         if found is not None:
             return found

scripts/compiler_differential/_helpers/comparing/comparison.py

@@ -19,6 +19,8 @@
     strip_report_fields,
 )
 from scripts.compiler_differential.constants import (
+    ANALYSIS_CACHE_STATE_POINTERS,
+    COLD_CAPTURE_COMMAND,
     MISSING_VALUE,
     MODEL_RESOURCE_TYPE,
     PLAN_LABEL,
@@ -266,7 +268,11 @@ def _capture_differences(
             differences.append(
                 (
                     artifact,
-                    first_capture_difference(left=left_files[name], right=right_files[name]),
+                    first_capture_difference(
+                        left=left_files[name],
+                        right=right_files[name],
+                        ignored=_cache_state_pointers(command),
+                    ),
                 )
             )
     return differences
@@ -278,3 +284,9 @@ def _presence_divergence(*, left_present: bool, right_present: bool) -> Divergen
         left="present" if left_present else MISSING_VALUE,
         right="present" if right_present else MISSING_VALUE,
     )
+
+
+def _cache_state_pointers(command: str) -> frozenset[str]:
+    """Catalog state Python's warm analysis cache skips; compared only on the cold compile."""
+
+    return frozenset() if command == COLD_CAPTURE_COMMAND else ANALYSIS_CACHE_STATE_POINTERS

scripts/compiler_differential/constants.py

@@ -86,6 +86,10 @@ def short_model_names(*, model: Model, ctx: RuleContext) -> list[Finding]:
 COLD_COMPILE: DifferentialCommand = DifferentialCommand(
     label="compile", arguments=("compile", "--json", "--manifest", "--dag")
 )
+COLD_CAPTURE_COMMAND: str = f"0-{COLD_COMPILE.label}"
+ANALYSIS_CACHE_STATE_POINTERS: frozenset[str] = frozenset(
+    {"/binding_catalog/schemas", "/binding_catalog/analysis_shapes"}
+)
 WARM_COMPILE: DifferentialCommand = DifferentialCommand(
     label="compile-warm", arguments=("compile", "--json")
 )

src/sqlbuild/compiler/analysis_session/classes/native_model_analysis.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+from collections import Counter
 from collections.abc import Sequence
 from typing import Any
 
@@ -62,6 +63,7 @@ def __init__(
             request.column_nullability_by_table
         )
         self._kept: dict[tuple[int, str], PolyglotAnalysisResult] = {}
+        self._deferrals: Counter[str] = Counter()
 
     def analyses(
         self, session: _native.NativeModelAnalysisSession
@@ -81,6 +83,8 @@ def analyses(
         self._record_catalog_changes(
             schema_additions=schema_additions, analysis_names=analysis_names
         )
+        for kind, count in self._deferrals.items():
+            record_analysis_deferral(kind=kind, count=count)
         self._python.record_uncached()
         return {
             request.model_input.model_file.file_path.stem: self._model_analysis(
@@ -123,7 +127,7 @@ def _answer(self, deferral: DeferralRow) -> DeferredRow:
         shapes: dict[str, dict[str, str]] = {name: dict(shape) for name, shape in schemas}
         analysis: PolyglotAnalysisResult
         if kind == DEFERRAL_ANALYSIS:
-            record_analysis_deferral(kind=DEFERRAL_LEGACY_ANALYSIS)
+            self._deferrals[DEFERRAL_LEGACY_ANALYSIS] += 1
             analysis = self._python.analyze_deferred(
                 model=model,
                 precomputed=PythonModelAnalysis.legacy_precomputed(
@@ -136,7 +140,7 @@ def _answer(self, deferral: DeferralRow) -> DeferredRow:
                 column_nullability_by_table=self._nullability,
             )
         else:
-            record_analysis_deferral(kind=DEFERRAL_ENRICHMENT)
+            self._deferrals[DEFERRAL_ENRICHMENT] += 1
             analysis = self._python.enrich(model=model, input_schemas=shapes)
         self._kept[(model, kind)] = analysis
         return deferred_row(model=model, analysis=analysis)

tests/e2e/scripts/compiler_differential/helpers.py

@@ -73,6 +73,20 @@ def _perturbed(*, until, **stages):
 frontier.native_frontier = _perturbed
 """
 
+CATALOG_PERTURBATION: str = """
+import sqlbuild.compiler.analysis_session.classes.native_model_analysis as native_analysis
+
+_original = native_analysis.NativeModelAnalysis._record_catalog_changes
+
+
+def _perturbed(self, **changes):
+    _original(self, **changes)
+    self._catalog.schemas["perturbed_by_test"] = {"order_id": "INTEGER"}
+
+
+native_analysis.NativeModelAnalysis._record_catalog_changes = _perturbed
+"""
+
 DEFERRAL_PERTURBATION: str = """
 import json
 import os

tests/e2e/scripts/compiler_differential/test_compiler_differential.py

@@ -18,6 +18,7 @@
     WheelSiteReportTestCase,
 )
 from tests.e2e.scripts.compiler_differential.helpers import (
+    CATALOG_PERTURBATION,
     COMPILED_PROJECT_CAPTURE,
     DEFERRAL_PERTURBATION,
     DISCOVERY_PERTURBATION,
@@ -181,6 +182,39 @@ def test_given_perturbed_native_render_when_comparing_then_compile_inputs_captur
     assert not any(text in output for text in test_case.expected_absent), output
 
 
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        HarnessRunTestCase(
+            description="wrong_native_catalog_shape",
+            extra_arguments=(),
+            expected_exit_code=1,
+            expected_lines=("DIFF project/waffle_shop",),
+            expected_patterns=(
+                r"- stage capture 0-compile/\d+-compiled_project\.json at /binding_catalog/schemas",
+            ),
+            expected_absent=("warm/", "3-plan/"),
+        )
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_wrong_native_catalog_shape_when_comparing_then_only_the_cold_capture_differs(
+    test_case: HarnessRunTestCase, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    exit_code: int = run_compiler_differential(
+        harness_arguments(
+            work_dir=tmp_path / "work",
+            extra=perturbation_arguments(tmp_path / "perturbation", source=CATALOG_PERTURBATION),
+        )
+    )
+
+    output: str = capsys.readouterr().out
+    assert exit_code == test_case.expected_exit_code, output
+    assert all(line in output for line in test_case.expected_lines), output
+    assert all(re.search(pattern, output) for pattern in test_case.expected_patterns), output
+    assert not any(text in output for text in test_case.expected_absent), output
+
+
 @pytest.mark.parametrize(
     "test_case",
     [

tests/integration/src/sqlbuild/compiler/analysis_session/_test_types.py

@@ -25,3 +25,13 @@ class AnalysisFallbackTestCase:
     allow_compact_analysis: bool
     keeps_catalog: bool
     expected_kind: str
+
+
+@dataclass(frozen=True)
+class SessionFailureTestCase:
+    """A session that fails after Python answered deferrals, so Python analyses everything."""
+
+    description: str
+    seed: int
+    model_count: int
+    expected_kinds: dict[str, int]

tests/integration/src/sqlbuild/compiler/analysis_session/helpers.py

@@ -13,6 +13,7 @@
 
 import pytest
 
+import sqlbuild._native as native_module
 import sqlbuild.compiler.compile._helpers.assembly.project as project_assembly
 from sqlbuild.adapter.contract.models import ExpressionInferenceProfile
 from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter
@@ -394,3 +395,36 @@ def _analysis_views(analyses: dict[str, ModelSqlAnalysis] | None) -> object:
 
 def _catalog_view(catalog: Any) -> tuple[object, ...]:
     return (dict(catalog.schemas), dict(catalog.analysis_shapes))
+
+
+class FailingProvideSession:
+    """A native session whose answers to deferrals are refused, as a failed session refuses."""
+
+    def __init__(self, session: Any) -> None:
+        self._session: Any = session
+        self.answered: int = 0
+        self.failure: str = "injected session failure"
+
+    def run(self) -> object:
+        return self._session.run()
+
+    def provide(self, answers: list[object]) -> bool:
+        self.answered += len(answers)
+        return False
+
+    def finish(self) -> object:
+        return self._session.finish()
+
+
+def failing_provide_sessions(*, monkeypatch: pytest.MonkeyPatch) -> list[FailingProvideSession]:
+    """Make every native session refuse Python's answers; return the sessions started."""
+
+    started: list[FailingProvideSession] = []
+    original: Callable[..., Any] = native_module.start_model_analysis_session
+
+    def start(catalog: object, request: tuple[object, ...]) -> FailingProvideSession:
+        started.append(FailingProvideSession(original(catalog, request)))
+        return started[-1]
+
+    monkeypatch.setattr(native_module, "start_model_analysis_session", start)
+    return started