perf(native): share equal inferred columns and guard custom rule CPU

.github/workflows/verify.yml

@@ -611,6 +611,12 @@ jobs:
             gate_phase: model_analysis_cpu_ms
             compile_args: --compile-arg=--select=*
             max_ratio: "1.50"
+          - kind: dense
+            label: native-preview custom rules vs python
+            base_engine: python
+            head_engine: native-preview
+            gate_phase: custom_rules_cpu_ms
+            max_ratio: "1.25"
     steps:
       - uses: actions/checkout@v4
       - name: Check out the comparison base

scripts/compile_performance_ratio/constants.py

@@ -53,6 +53,7 @@
     "contracts_cpu_ms",
     "built_in_rules_ms",
     "custom_rules_ms",
+    "custom_rules_cpu_ms",
     "test_planning_ms",
     "write_ms",
 )

src/sqlbuild/cli/commands/_helpers/compile/pipeline.py

@@ -260,6 +260,7 @@ def _analyze_compile_project(
         contract_cpu_ms=contract_cpu_ms,
         built_in_rules_ms=rules_result.built_in_ms,
         custom_rules_ms=rules_result.custom_ms,
+        custom_rules_cpu_ms=rules_result.custom_cpu_ms,
         early_lint_wait_ms=early_lint_wait_ms,
         rule_cache_hits=rules_result.cache_hits,
         rule_cache_misses=rules_result.cache_misses,

src/sqlbuild/cli/commands/main/project/_compile.py

@@ -169,6 +169,7 @@ def _run_compile_with_status(
         "contracts_cpu_ms": analysis.contract_cpu_ms,
         "built_in_rules_ms": analysis.built_in_rules_ms,
         "custom_rules_ms": analysis.custom_rules_ms,
+        "custom_rules_cpu_ms": analysis.custom_rules_cpu_ms,
         "early_lint_wait_ms": analysis.early_lint_wait_ms,
         "rule_cache_hits": analysis.rule_cache_hits,
         "rule_cache_misses": analysis.rule_cache_misses,

src/sqlbuild/cli/compile/models.py

@@ -45,6 +45,7 @@ class CompileAnalysis:
     contract_cpu_ms: int
     built_in_rules_ms: int = 0
     custom_rules_ms: int = 0
+    custom_rules_cpu_ms: int = 0
     early_lint_wait_ms: int = 0
     rule_cache_hits: int = 0
     rule_cache_misses: int = 0

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

@@ -14,7 +14,6 @@
     compact_lineage,
     contract_proof,
     deferred_row,
-    inferred_columns,
     lineage_facts,
 )
 from sqlbuild.compiler.analysis_session.constants import (
@@ -31,6 +30,7 @@
 )
 from sqlbuild.compiler.analysis_session.models import NativeModelAnalysisRequest
 from sqlbuild.compiler.analysis_session.types import (
+    ColumnRow,
     DeferralRow,
     DeferredRow,
     FinishRow,
@@ -42,6 +42,7 @@
 from sqlbuild.compiler.compile.classes.python_model_analysis import PythonModelAnalysis
 from sqlbuild.compiler.compile.models import (
     CompiledLineageColumnFact,
+    InferredColumn,
     ModelSqlAnalysis,
     ModelSqlAnalysisRequest,
     PolyglotAnalysisResult,
@@ -69,6 +70,7 @@ def __init__(
             request.column_nullability_by_table
         )
         self._kept: dict[tuple[int, str], PolyglotAnalysisResult] = {}
+        self._columns: dict[ColumnRow, InferredColumn] = {}
         self._deferrals: Counter[str] = Counter()
 
     def analyses(
@@ -176,6 +178,23 @@ def _record_catalog_changes(
             },
         )
 
+    def _inferred_columns(self, rows: list[ColumnRow] | None) -> tuple[InferredColumn, ...] | None:
+        """Columns as Python's analysis shares them: one object per distinct column value."""
+
+        if rows is None:
+            return None
+        return tuple(map(self._inferred_column, rows))
+
+    def _inferred_column(self, row: ColumnRow) -> InferredColumn:
+        column: InferredColumn | None = self._columns.get(row)
+        if column is None:
+            name, data_type, nullability = row
+            column = InferredColumn(
+                name=name, type=data_type, nullability=InferredNullability(nullability)
+            )
+            self._columns[row] = column
+        return column
+
     def _model_analysis(
         self,
         *,
@@ -205,7 +224,7 @@ def _model_analysis(
         return ModelSqlAnalysis(
             polyglot_analysis=PolyglotAnalysisResult(
                 analysis_succeeded=succeeded,
-                columns=inferred_columns(columns),
+                columns=self._inferred_columns(columns),
                 lineage_columns=lineage,
                 has_star=has_star,
                 star_resolved=star_resolved,

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

@@ -80,6 +80,7 @@ def evaluate_custom_rules_cached(
     """Reuse results whose recorded fact reads are unchanged and evaluate only the rest."""
 
     started: float = time.monotonic()
+    started_cpu: float = _cpu_seconds()
     custom: tuple[Rule, ...] = tuple(sorted(rules, key=lambda rule: rule.code))
     if not custom:
         return CustomRulesOutcome(findings=(), cache_hits=0, cache_misses=0, custom_ms=0)
@@ -105,6 +106,7 @@ def evaluate_custom_rules_cached(
             cache_hits=0,
             cache_misses=len(custom),
             custom_ms=_elapsed_ms(started),
+            custom_cpu_ms=_cpu_ms(started_cpu),
         )
     digests: FactDigests = FactDigests(
         views=_views_factory(project=facts_project, project_dir=project_dir, dialect=dialect)
@@ -185,6 +187,7 @@ def evaluate_custom_rules_cached(
         cache_hits=hits,
         cache_misses=misses,
         custom_ms=_elapsed_ms(started),
+        custom_cpu_ms=_cpu_ms(started_cpu),
     )
 
 
@@ -476,5 +479,16 @@ def _digest(value: object) -> str:
     return hashlib.sha256(orjson.dumps(value, default=str)).hexdigest()
 
 
+def _cpu_seconds() -> float:
+    """This thread's CPU plus that of finished child processes, which run the rule hosts."""
+
+    times: os.times_result = os.times()
+    return time.thread_time() + times.children_user + times.children_system
+
+
+def _cpu_ms(started: float) -> int:
+    return int((_cpu_seconds() - started) * 1000)
+
+
 def _elapsed_ms(started: float) -> int:
     return round((time.monotonic() - started) * 1000)

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

@@ -161,6 +161,7 @@ def evaluate_native(
         cache_misses=(0 if reused else native_misses) + custom.cache_misses,
         built_in_ms=0 if reused else int(payload.get("built_in_ms", 0)),
         custom_ms=custom.custom_ms,
+        custom_cpu_ms=custom.custom_cpu_ms,
     )
 
 

src/sqlbuild/rule_engine/_helpers/run/rules.py

@@ -231,6 +231,7 @@ def evaluate_rules(
         unevaluated_resources=len({item.path for item in findings if item.unevaluated}),
         built_in_ms=sql_ms + result.built_in_ms,
         custom_ms=result.custom_ms,
+        custom_cpu_ms=result.custom_cpu_ms,
         cache_hits=result.cache_hits + sql_result.cache_hits,
         cache_misses=result.cache_misses + sql_result.cache_misses,
         skipped_type_proof_rules=skipped_type_proof_rules,

src/sqlbuild/rule_engine/models.py

@@ -311,6 +311,7 @@ class RulesResult:
     cache_misses: int = 0
     built_in_ms: int = 0
     custom_ms: int = 0
+    custom_cpu_ms: int = 0
 
 
 @dataclass(frozen=True)
@@ -346,6 +347,7 @@ class RulesRunResult:
     evaluated_models: int
     built_in_ms: int
     custom_ms: int
+    custom_cpu_ms: int = 0
     unevaluated_resources: int = 0
     cache_hits: int = 0
     cache_misses: int = 0
@@ -431,6 +433,7 @@ class CustomRulesOutcome:
     cache_hits: int
     cache_misses: int
     custom_ms: int
+    custom_cpu_ms: int = 0
 
 
 @dataclass(frozen=True)

tests/e2e/scripts/compile_performance_ratio/test_compare_compile_performance.py

@@ -206,6 +206,18 @@ def test_given_base_root_when_comparing_compile_performance_then_each_side_uses_
             gate_phase="model_analysis_cpu_ms",
             compile_args=("--compile-arg=--select=*",),
         ),
+        EnginePhaseRatioTestCase(
+            description="custom rule CPU over its allowance fails on it alone",
+            max_ratio="0",
+            phase_noise_floor_ms="-1",
+            expected_return_code=1,
+            expected_fragments=(
+                "Gated phases: custom_rules_cpu_ms.",
+                "| custom_rules_cpu_ms |",
+                "Compile performance regression: dense 20 cold custom_rules_cpu_ms:",
+            ),
+            gate_phase="custom_rules_cpu_ms",
+        ),
     ),
     ids=lambda case: case.description,
 )

tests/e2e/src/sqlbuild/cli/commands/main/build/test_compile_json_behavior.py

@@ -71,6 +71,7 @@ def test_given_waffle_shop_when_running_compile_json_then_it_reports_offline_que
         "attachment_ms",
         "built_in_rules_ms",
         "custom_rules_ms",
+        "custom_rules_cpu_ms",
         "early_lint_wait_ms",
         "model_analysis_ms",
         "model_analysis_cpu_ms",

tests/e2e/src/sqlbuild/cli/commands/main/compile/test_dense_compile_performance.py

@@ -125,6 +125,7 @@ def test_given_dense_project_when_compiling_cold_then_preserves_rules_semantics_
     assert timings["rule_cache_misses"] == 2 * test_case.model_count + 1
     assert timings["built_in_rules_ms"] > 0
     assert timings["custom_rules_ms"] > 0
+    assert timings["custom_rules_cpu_ms"] > 0
     assert result.semantic_fingerprint == test_case.expected_fingerprint
     assert result.elapsed_seconds < test_case.expected_max_wall_seconds
     assert result.peak_rss_bytes < test_case.expected_max_rss_bytes

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

@@ -63,3 +63,4 @@ class SharedAnalysisTestCase:
     expected_shared: int
     expected_reanalysed: int
     expected_unshared: int
+    expected_column_values: int

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

@@ -40,6 +40,7 @@
     CompiledLineageColumnFact,
     CompileProjectInputs,
     DynamicColumnContractProof,
+    InferredColumn,
     ModelSqlAnalysis,
 )
 from sqlbuild.compiler.discovery.main.discover import discover_project_inputs
@@ -430,6 +431,8 @@ class AnalysisParity:
     standalone_proofs: int = 0
     proven_pivots: int = 0
     native_enrichments: int = 0
+    native_column_objects: int = 0
+    native_column_values: int = 0
 
 
 def compare_analyses(
@@ -459,6 +462,11 @@ def models_by_both(
             )
         )
         python: dict[str, ModelSqlAnalysis] = python_analysis.func(**keywords)
+        columns: list[InferredColumn] = list(
+            chain.from_iterable(map(_native_columns, (native or {}).values()))
+        )
+        parity.native_column_objects += len(set(map(id, columns)))
+        parity.native_column_values += len(set(columns))
         parity.analysed_models += len(python)
         _append(parity, "model analyses", _analysis_views(python), _analysis_views(native))
         _append(
@@ -569,6 +577,10 @@ def deferral_kinds(record_dir: Path) -> Counter[str]:
     return Counter(f"{record['site']}:{record['kind']}" for record in records)
 
 
+def _native_columns(analysis: ModelSqlAnalysis) -> tuple[InferredColumn, ...]:
+    return analysis.polyglot_analysis.columns or ()
+
+
 def _append(parity: AnalysisParity, name: str, python: object, native: object) -> None:
     parity.names.append(name)
     parity.python.append(python)

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

@@ -107,6 +107,7 @@ def test_given_generated_projects_when_analysing_natively_then_matches_python(
         <= test_case.expected_maximum_enrichment_deferrals
     )
     assert parity.native_enrichments >= test_case.expected_minimum_native_enrichments
+    assert parity.native_column_objects == parity.native_column_values > 0
 
 
 @pytest.mark.parametrize(
@@ -121,6 +122,7 @@ def test_given_generated_projects_when_analysing_natively_then_matches_python(
             expected_shared=22,
             expected_reanalysed=6,
             expected_unshared=2,
+            expected_column_values=12,
         )
     ],
     ids=lambda case: case.description,
@@ -153,12 +155,16 @@ def test_given_equal_model_queries_when_analysing_natively_then_shares_and_match
         sum(session.sharing[1] for session in sessions),
         parity.analysed_models - shared,
         deferral_kinds(record_dir),
+        parity.native_column_objects,
+        parity.native_column_values,
     ) == (
         test_case.expected_analysed,
         test_case.expected_shared,
         test_case.expected_reanalysed,
         test_case.expected_unshared,
         Counter(),
+        test_case.expected_column_values,
+        test_case.expected_column_values,
     )