perf(compile): stop the unused early SQL lint and map lint lines once per file (#919)

Signal the speculative lint thread to stop as soon as the compile knows its
result will not be consumed (graph errors or end of analysis), checking the
stop between files and bodies instead of joining a full lint run. Report any
remaining join as compile_timings.early_lint_wait_ms.

Build each file's newline table once per native lint pass instead of once per
violation.

Author: kvlonge

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)