perf(compile): stop the speculative SQL lint when its result is unused

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

@@ -155,6 +155,8 @@ def _analyze_compile_project(
     _ = complete_compile_phase(
         status=status, message=f"Compiled project graph. ({graph_ms / 1000:.2f}s)"
     )
+    graph_failed: bool = any(diagnostic.is_error for diagnostic in graph.project.diagnostics)
+    early_lint_wait_ms: int = early_lint.stop() if graph_failed else 0
     lineage_start: float = time.monotonic()
     _ = start_compile_phase(status=status, message="Analyzing column lineage...")
     lineage: ProjectColumnLineage | None = build_compile_lineage(
@@ -197,7 +199,7 @@ def _analyze_compile_project(
         *graph.project.diagnostics,
         *contract_result.diagnostics,
     )
-    if not any(diagnostic.is_error for diagnostic in graph.project.diagnostics):
+    if not graph_failed:
         rules_config: RulesConfig = load_rules_config(project_dir=project_dir)
         if (
             prepared_artifacts is not None
@@ -224,6 +226,7 @@ def _analyze_compile_project(
                 elapsed_seconds=time.monotonic() - rules_start, result=rules_result
             ),
         )
+        early_lint_wait_ms = early_lint.stop()
     rule_diagnostics: tuple[CompilerDiagnostic, ...] = tuple(
         CompilerDiagnostic(
             phase=DiagnosticPhase.RULE,
@@ -252,6 +255,7 @@ def _analyze_compile_project(
         contract_ms=contract_ms,
         built_in_rules_ms=rules_result.built_in_ms,
         custom_rules_ms=rules_result.custom_ms,
+        early_lint_wait_ms=early_lint_wait_ms,
         rule_cache_hits=rules_result.cache_hits,
         rule_cache_misses=rules_result.cache_misses,
         skipped_type_proof_rules=rules_result.skipped_type_proof_rules,

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

@@ -124,6 +124,7 @@ def _run_compile_with_status(
         "contracts_ms": analysis.contract_ms,
         "built_in_rules_ms": analysis.built_in_rules_ms,
         "custom_rules_ms": analysis.custom_rules_ms,
+        "early_lint_wait_ms": analysis.early_lint_wait_ms,
         "rule_cache_hits": analysis.rule_cache_hits,
         "rule_cache_misses": analysis.rule_cache_misses,
         "write_ms": write_result.write_ms,

src/sqlbuild/cli/compile/models.py

@@ -60,6 +60,7 @@ class CompileAnalysis:
     contract_ms: int
     built_in_rules_ms: int = 0
     custom_rules_ms: int = 0
+    early_lint_wait_ms: int = 0
     rule_cache_hits: int = 0
     rule_cache_misses: int = 0
     skipped_type_proof_rules: tuple[str, ...] = ()

src/sqlbuild/lint/_helpers/native_sql.py

@@ -5,6 +5,7 @@
 import re
 from bisect import bisect_right
 from pathlib import Path
+from threading import Event
 from typing import Any, cast
 
 import orjson
@@ -28,14 +29,15 @@
 from sqlbuild.compiler.compile.main.map_expanded_offset import map_expanded_offset
 from sqlbuild.compiler.compile.models import MappedOffset
 from sqlbuild.lint._helpers.sqlbuild_tokens import interpolation_text_at
+from sqlbuild.lint.classes.stop_context import LintStopContext
 from sqlbuild.lint.constants import (
     GENERATED_SQL_MESSAGE_SUFFIX,
     GENERATED_SQL_MESSAGE_TEMPLATE,
     LINT_ENGINE_NATIVE,
     VIOLATION_SEVERITY_FAULT,
     VIOLATION_SEVERITY_WARNING,
 )
-from sqlbuild.lint.exceptions import NativeLintError
+from sqlbuild.lint.exceptions import LintStoppedError, NativeLintError
 from sqlbuild.lint.models import LintBody, LintConfig, LintEdit, LintViolation
 
 _NATIVE_LINT_API_VERSION: int = 1
@@ -87,6 +89,7 @@ def run_native_sql_lint(
     """Lint expanded bodies in Rust and map diagnostics to authored source."""
 
     violations_by_file: dict[Path, list[LintViolation]] = {}
+    line_starts_by_path: dict[Path, tuple[int, ...]] = {}
     requests: dict[_NativeCacheKey, dict[str, object]] = {}
     for body in bodies:
         cache_key: _NativeCacheKey = _native_cache_key(body=body, config=config)
@@ -118,15 +121,20 @@ def run_native_sql_lint(
         if cache_key[-1]:
             payload["relation_keys"] = dict(cache_key[-1])
         requests[cache_key] = payload
+    check_lint_not_stopped()
     response_cache: dict[_NativeCacheKey, _NativeResult] = _native_responses(requests=requests)
 
     for body in bodies:
+        check_lint_not_stopped()
         response: _NativeResult = response_cache[_native_cache_key(body=body, config=config)]
+        contents: str = contents_by_path[body.file_path]
+        line_starts: tuple[int, ...] | None = line_starts_by_path.get(body.file_path)
+        if line_starts is None:
+            line_starts = _line_starts(contents)
+            line_starts_by_path[body.file_path] = line_starts
         if isinstance(response, NativeLintError):
             parse_violation: LintViolation | None = _parse_failure_violation(
-                error=response,
-                body=body,
-                contents=contents_by_path[body.file_path],
+                error=response, body=body, line_starts=line_starts
             )
             if parse_violation is None:
                 raise response
@@ -140,18 +148,27 @@ def run_native_sql_lint(
                 violation: LintViolation | None = _authored_violation(
                     raw_diagnostic=raw_diagnostic,
                     body=body,
-                    contents=contents_by_path[body.file_path],
+                    contents=contents,
+                    line_starts=line_starts,
                     dialect=config.dialect,
                 )
             except NativeLintError as error:
                 violation = _parse_failure_violation(
-                    error=error, body=body, contents=contents_by_path[body.file_path]
+                    error=error, body=body, line_starts=line_starts
                 )
             if violation is not None:
                 violations_by_file.setdefault(body.file_path, []).append(violation)
     return {path: tuple(entries) for path, entries in violations_by_file.items()}
 
 
+def check_lint_not_stopped() -> None:
+    """Abandon a speculative lint run on this thread once its owner no longer needs it."""
+
+    stop: Event | None = LintStopContext.active.get()
+    if stop is not None and stop.is_set():
+        raise LintStoppedError
+
+
 def _native_cache_key(*, body: LintBody, config: LintConfig) -> _NativeCacheKey:
     return (
         body.lint_text,
@@ -223,7 +240,7 @@ def _native_result(*, response: object) -> _NativeResult:
 
 
 def _parse_failure_violation(
-    *, error: NativeLintError, body: LintBody, contents: str
+    *, error: NativeLintError, body: LintBody, line_starts: tuple[int, ...]
 ) -> LintViolation:
     """Convert a per-body native parse rejection into an authored project fault."""
 
@@ -237,7 +254,7 @@ def _parse_failure_violation(
         expanded_offset += min(expanded_column - 1, len(expanded_lines[expanded_line - 1]))
     mapped: MappedOffset = map_expanded_offset(offset=expanded_offset, passes=body.passes)
     absolute_offset: int = body.body_start + mapped.offset
-    line, column = _offset_position(offset=absolute_offset, line_starts=_line_starts(contents))
+    line, column = _offset_position(offset=absolute_offset, line_starts=line_starts)
     return LintViolation(
         file_path=body.file_path,
         line=line,
@@ -251,7 +268,12 @@ def _parse_failure_violation(
 
 
 def _authored_violation(
-    *, raw_diagnostic: object, body: LintBody, contents: str, dialect: str
+    *,
+    raw_diagnostic: object,
+    body: LintBody,
+    contents: str,
+    line_starts: tuple[int, ...],
+    dialect: str,
 ) -> LintViolation | None:
     if not isinstance(raw_diagnostic, dict):
         raise NativeLintError("native lint diagnostic must be an object")
@@ -286,16 +308,15 @@ def _authored_violation(
         _SQLBUILD_HARNESS_CTE_PREFIXES, absolute_offset
     ):
         return None
-    starts: tuple[int, ...] = _line_starts(contents)
-    line, column = _offset_position(offset=absolute_offset, line_starts=starts)
+    line, column = _offset_position(offset=absolute_offset, line_starts=line_starts)
     end_position: tuple[int, int] | None = _authored_end_position(
         start=start,
         end=end,
         mapped_start=mapped,
         absolute_start=absolute_offset,
         body=body,
         contents=contents,
-        line_starts=starts,
+        line_starts=line_starts,
         dialect=dialect,
     )
     fix: LintEdit | None = _authored_fix(

src/sqlbuild/lint/classes/__init__.py

@@ -0,0 +1 @@
+"""Runtime classes for the lint and format layer."""

src/sqlbuild/lint/classes/stop_context.py

@@ -0,0 +1,11 @@
+"""Thread-local stop signal for speculative lint runs."""
+
+from contextvars import ContextVar
+from threading import Event
+from typing import ClassVar
+
+
+class LintStopContext:
+    """Own the stop signal that lets a speculative lint run be abandoned between files."""
+
+    active: ClassVar[ContextVar[Event | None]] = ContextVar("active_lint_stop", default=None)

src/sqlbuild/lint/exceptions.py

@@ -31,3 +31,7 @@ class ProjectCompileError(LintError):
     """The project could not be compiled, so its SQL cannot be linted."""
 
     code: str = "L005"
+
+
+class LintStoppedError(Exception):
+    """A speculative lint run was stopped because its result is no longer needed."""

src/sqlbuild/lint/main/run_lint.py

@@ -17,7 +17,7 @@
 from sqlbuild.lint._helpers.expansion import build_lint_expansion_context, prepare_lint_body
 from sqlbuild.lint._helpers.headers import lint_body_ranges, lint_file_role, scan_headers
 from sqlbuild.lint._helpers.native import external_identifiers_for_headers, lint_native_headers
-from sqlbuild.lint._helpers.native_sql import run_native_sql_lint
+from sqlbuild.lint._helpers.native_sql import check_lint_not_stopped, run_native_sql_lint
 from sqlbuild.lint._helpers.project_files import collect_project_files, sort_violations
 from sqlbuild.lint._helpers.suppressions import apply_suppressions
 from sqlbuild.lint.constants import HEADER_KIND_SCENARIO, HEADER_KIND_TEST
@@ -154,6 +154,7 @@ def _prepared_bodies(
 ) -> tuple[LintBody, ...] | str:
     """Prepared bodies of one file, or the reason its SQL cannot be expanded when skipping."""
 
+    check_lint_not_stopped()
     bodies: list[LintBody] = []
     role: LintFileRole = lint_file_role(
         file_path=file_path, project_dir=project_dir, relative_path=relative_path

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

@@ -7,8 +7,10 @@
 import os
 import time
 from concurrent.futures import Executor, Future, ThreadPoolExecutor
+from contextvars import Token
 from dataclasses import asdict, dataclass, replace
 from pathlib import Path
+from threading import Event
 from typing import Any
 
 import sqlbuild._native as _native
@@ -25,6 +27,7 @@
 from sqlbuild.compiler.compile.types import CompiledResourceType
 from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs
 from sqlbuild.compiler.pipeline.models import ProjectGraph
+from sqlbuild.lint.classes.stop_context import LintStopContext
 from sqlbuild.lint.constants import (
     DEFAULT_MAX_LITERAL_LENGTH,
     DEFAULT_MAX_RANKING_ORDER_BY,
@@ -244,7 +247,7 @@ def _config_references_custom_rules(config: RulesConfig) -> bool:
 
 
 def prepare_sql_rules(
-    *, inputs: CompileProjectInputs, executor: Executor, dialect: str
+    *, inputs: CompileProjectInputs, executor: Executor, dialect: str, stop: Event
 ) -> PreparedSqlLint | None:
     project_dir: Path | None = inputs.discovered_inputs.project_dir
     if project_dir is None or any(diagnostic.is_error for diagnostic in inputs.diagnostics):
@@ -302,6 +305,7 @@ def prepare_sql_rules(
                 for model in inputs.model_inputs
                 if model.sql_expansion is not None
             },
+            stop=stop,
         ),
     )
 
@@ -313,19 +317,24 @@ def _prepare_sql_lint(
     discovered_inputs: DiscoveredProjectInputs,
     static_declaration_scope: DeclarationScopeBuild | None,
     compiled_expansions: dict[Path, CompiledSqlExpansion],
+    stop: Event,
 ) -> PreparedSqlLintResult:
-    context: SqlExpansionContext = build_expansion_context(
-        project_dir=project_dir,
-        discovered_inputs=discovered_inputs,
-        static_declaration_scope=static_declaration_scope,
-    )
-    result: LintRunResult = run_lint(
-        project_dir=project_dir,
-        config=config,
-        discovered_inputs=discovered_inputs,
-        compiled_expansions=compiled_expansions,
-        expansion_context=context,
-    )
+    stop_token: Token[Event | None] = LintStopContext.active.set(stop)
+    try:
+        context: SqlExpansionContext = build_expansion_context(
+            project_dir=project_dir,
+            discovered_inputs=discovered_inputs,
+            static_declaration_scope=static_declaration_scope,
+        )
+        result: LintRunResult = run_lint(
+            project_dir=project_dir,
+            config=config,
+            discovered_inputs=discovered_inputs,
+            compiled_expansions=compiled_expansions,
+            expansion_context=context,
+        )
+    finally:
+        LintStopContext.active.reset(stop_token)
     return PreparedSqlLintResult(result=result, context=context)
 
 

src/sqlbuild/rule_engine/classes/early_sql_lint.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import time
 from concurrent.futures import ThreadPoolExecutor
+from threading import Event
 from types import TracebackType
 
 from sqlbuild.compiler.compile.models import CompileProjectInputs
@@ -15,6 +17,7 @@ def __init__(self, *, enabled: bool) -> None:
         self.enabled: bool = enabled
         self.preparation: PreparedSqlLint | None = None
         self.expansion_reuse: SqlExpansionReuse | None = None
+        self.stop_requested: Event = Event()
         self.executor: ThreadPoolExecutor = ThreadPoolExecutor(
             max_workers=1, thread_name_prefix="sqlbuild-lint"
         )
@@ -28,12 +31,15 @@ def __exit__(
         exception: BaseException | None,
         traceback: TracebackType | None,
     ) -> None:
-        self.executor.shutdown(wait=True, cancel_futures=True)
+        _ = self.stop()
 
     def start(self, *, inputs: CompileProjectInputs, dialect: str) -> None:
         if self.enabled:
             self.preparation = prepare_sql_rules(
-                inputs=inputs, executor=self.executor, dialect=dialect
+                inputs=inputs,
+                executor=self.executor,
+                dialect=dialect,
+                stop=self.stop_requested,
             )
             self.expansion_reuse = (
                 None
@@ -43,3 +49,11 @@ def start(self, *, inputs: CompileProjectInputs, dialect: str) -> None:
                     declaration_scope=inputs.declaration_scope,
                 )
             )
+
+    def stop(self) -> int:
+        """Abandon unconsumed speculative work and return the milliseconds spent joining it."""
+
+        stop_start: float = time.monotonic()
+        self.stop_requested.set()
+        self.executor.shutdown(wait=True, cancel_futures=True)
+        return int((time.monotonic() - stop_start) * 1000)

src/sqlbuild/rule_engine/main/_prepare_sql_rules.py

@@ -3,15 +3,16 @@
 from __future__ import annotations
 
 from concurrent.futures import Executor
+from threading import Event
 
 from sqlbuild.compiler.compile.models import CompileProjectInputs
 from sqlbuild.rule_engine._helpers.run.rules import prepare_sql_rules as prepare
 from sqlbuild.rule_engine.models import PreparedSqlLint
 
 
 def prepare_sql_rules(
-    *, inputs: CompileProjectInputs, executor: Executor, dialect: str
+    *, inputs: CompileProjectInputs, executor: Executor, dialect: str, stop: Event
 ) -> PreparedSqlLint | None:
     """Prepare configured SQL checks without publishing findings before compilation finishes."""
 
-    return prepare(inputs=inputs, executor=executor, dialect=dialect)
+    return prepare(inputs=inputs, executor=executor, dialect=dialect, stop=stop)

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

@@ -70,6 +70,7 @@ def test_given_waffle_shop_when_running_compile_json_then_it_reports_offline_que
         "attachment_ms",
         "built_in_rules_ms",
         "custom_rules_ms",
+        "early_lint_wait_ms",
         "model_analysis_ms",
         "analysis_preparation_ms",
         "analysis_native_ms",

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

@@ -539,3 +539,13 @@ class FormatDescriptionResolutionE2ECase:
     extra_files: tuple[tuple[str, str], ...]
     expected_returncode: int
     expected_faults: tuple[tuple[str, str], ...]
+
+
+@dataclass(frozen=True)
+class EarlyLintCompileTestCase:
+    """A rules-enabled project and the diagnostics compile must report for it."""
+
+    description: str
+    order_totals_sql: str
+    expected_exit_code: int
+    expected_diagnostics: tuple[tuple[str, str, int, int], ...]