fix: align configured rules with selected analysis (#752)

Co-authored-by: Kevin Longe <273747162+kvlonge@users.noreply.github.com>

Author: kvlonge

src/sqlbuild/compiler/pipeline/_helpers/analysis_selection.py

@@ -2,9 +2,21 @@
 
 from __future__ import annotations
 
-from sqlbuild.compiler.compile.models import CompileAnalysisSelection
+from sqlbuild.compiler.compile.models import (
+    CompileAnalysisSelection,
+    CompiledObjectKey,
+    CompiledProject,
+)
 from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs
 from sqlbuild.compiler.pipeline.models import CompilePipelineOptions
+from sqlbuild.compiler.planner.main.selection._resolve_planner_scopes import (
+    resolve_planner_scopes,
+)
+from sqlbuild.compiler.planner.models import (
+    PlannerPolicies,
+    PlannerScopeResolution,
+    PlannerSelection,
+)
 
 
 def resolve_compile_analysis_selection(
@@ -28,3 +40,30 @@ def resolve_compile_analysis_selection(
         auto_load_sources=options.auto_load_sources,
         no_cache=options.no_cache,
     )
+
+
+def resolve_configured_rule_selection(
+    *,
+    options: CompilePipelineOptions,
+    discovered_inputs: DiscoveredProjectInputs,
+    project: CompiledProject,
+) -> frozenset[CompiledObjectKey] | None:
+    """Align configured-rule subjects with models that received deep SQL analysis."""
+
+    analysis_selection: CompileAnalysisSelection | None = resolve_compile_analysis_selection(
+        options=options,
+        discovered_inputs=discovered_inputs,
+    )
+    if analysis_selection is None or (
+        not analysis_selection.select and not analysis_selection.exclude
+    ):
+        return None
+    scopes: PlannerScopeResolution = resolve_planner_scopes(
+        project=project,
+        selection=PlannerSelection(
+            select=analysis_selection.select,
+            exclude=analysis_selection.exclude,
+        ),
+        policies=PlannerPolicies(auto_load_sources=analysis_selection.auto_load_sources),
+    )
+    return frozenset(model.key for model in scopes.stale_warning_scope.models_by_name.values())

src/sqlbuild/compiler/pipeline/main/compile.py

@@ -22,6 +22,9 @@
 from sqlbuild.compiler.graph.main._build_lineage_upstream_deps import (
     build_lineage_upstream_deps,
 )
+from sqlbuild.compiler.pipeline._helpers.analysis_selection import (
+    resolve_configured_rule_selection,
+)
 from sqlbuild.compiler.pipeline._helpers.deferred_locations import (
     build_deferred_locations,
     gather_deferred_relations,
@@ -106,7 +109,11 @@ def run_compile_pipeline(
         config=rules_config,
         project_dir=project_dir,
         dialect=adapter.sql_analysis_dialect() or "generic",
-        selected_keys=None,
+        selected_keys=resolve_configured_rule_selection(
+            options=resolved_options,
+            discovered_inputs=discovered_inputs,
+            project=project,
+        ),
     )
     if on_progress is not None:
         on_progress(

tests/integration/src/sqlbuild/cli/commands/main/test_rules.py

@@ -1687,6 +1687,66 @@ def test_given_exception_for_other_model_when_compiling_selection_then_exception
     assert payload["diagnostics"] == []
 
 
+@pytest.mark.parametrize(
+    "test_case",
+    [RulesIntegrationTestCase("typed unselected model stays outside rule scope", 0, "")],
+    ids=lambda case: case.description,
+)
+def test_given_typed_unselected_model_when_planning_selection_then_rules_use_selected_scope(
+    test_case: RulesIntegrationTestCase,
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    (tmp_path / "sqlbuild_project.toml").write_text(
+        'name = "orders"\nadapter = "duckdb"\n\n[rules]\nselect = ["SQBRCONTRACT105"]\n',
+        encoding="utf-8",
+    )
+    sources: Path = tmp_path / "sources"
+    sources.mkdir()
+    (sources / "customers.yml").write_text(
+        """sources:
+  - name: raw_customers
+    schema: raw
+    table: customers
+    columns:
+      - name: customer_id
+        type: VARCHAR
+""",
+        encoding="utf-8",
+    )
+    models: Path = tmp_path / "models"
+    models.mkdir()
+    (models / "orders.sql").write_text(
+        "MODEL (contract enforced, columns (order_id (type INTEGER)));\n"
+        "SELECT CAST(1 AS INTEGER) AS order_id\n",
+        encoding="utf-8",
+    )
+    (models / "customers.sql").write_text(
+        """MODEL (
+  contract enforced,
+  columns (customer_id (type VARCHAR)),
+);
+
+WITH source_rows AS (
+  SELECT customer_id FROM __source("raw_customers")
+), final AS (
+  SELECT customer_id FROM source_rows
+)
+SELECT customer_id FROM final
+""",
+        encoding="utf-8",
+    )
+
+    exit_code: int = main(["--project-dir", str(tmp_path), "plan", "--json", "--select", "orders"])
+    payload: dict[str, object] = json.loads(capsys.readouterr().out)
+
+    assert exit_code == test_case.expected_exit_code
+    assert payload["selected_count"] == 1
+    assert [model["name"] for model in cast(list[dict[str, object]], payload["models"])] == [
+        "orders"
+    ]
+
+
 @pytest.mark.parametrize(
     "test_case",
     [RulesIntegrationTestCase("subject folder precedes model layer", 1, "SQBRPROJECT102")],

tests/unit/src/sqlbuild/compiler/pipeline/main/test_compile_phase_timings.py

@@ -54,6 +54,9 @@ def test_given_slow_planning_connection_when_compiling_then_direct_phases_are_di
     monkeypatch.setattr(compile_module, "open_connection_with_hooks", Mock(return_value=object()))
     monkeypatch.setattr(compile_module, "_build_result", Mock(return_value=pipeline_result))
     monkeypatch.setattr(compile_module, "load_rules_config", Mock(return_value=RulesConfig()))
+    monkeypatch.setattr(
+        compile_module, "resolve_configured_rule_selection", Mock(return_value=None)
+    )
     monkeypatch.setattr(
         compile_module,
         "run_rules",