perf(compile): reuse the previous compile when no input changed

src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/compile.md

@@ -12,6 +12,7 @@ Online: https://sqlbuild.com/docs/cli/compile/
 - Flags
 - What compile does
 - Focused compilation
+- Reusing an unchanged compile
 - Static analysis
 - Output
 - Column lineage modes
@@ -69,6 +70,78 @@ with selected model, seed, and function counts.
 Manifest and DAG outputs remain full-project artifacts. Focused compilation changes analysis and
 reporting scope; it does not produce a partial project graph.
 
+## Reusing an unchanged compile
+
+When nothing that can change the result has changed since the last compile, `sqb compile` replays
+that result instead of compiling again. It prints one line to stderr, for example
+`Inputs unchanged; reused the previous compile (0.1 s)`, then the same report and exit code as the
+previous run. Failed compiles are replayed too, with the same diagnostics.
+
+A compile is reused only when all of these match the stored run:
+
+- Every file and directory under the project, compared by type, size, modification and change
+  times, and inode. When those differ, SQLBuild compares content only for files whose content
+  digest it already knows; otherwise the compile runs in full. Digests are recorded for files that
+  changed within two seconds of a compile and for files whose timestamps moved since the previous
+  compile, so large unchanged files are never read. After a checkout that only moves timestamps,
+  the next compile runs in full once, and later timestamp-only changes are reused.
+- The command line: project directory, working directory, flags, selectors, `--vars`, target, and
+  whether color output is active.
+- Environment variables that SQLBuild reads: every `SQLBUILD_*` and `SQB_*` variable, every
+  variable that a template or `@@ENV` read during the stored compile, and every variable a
+  provider's settings can read, including `env_prefix`, aliases, and nested-delimiter variables.
+  Values are stored only as hashes.
+- The contents of each provider settings `env_file` and `secrets_dir`.
+- The Python interpreter, the SQLBuild version and native extension, `sys.path` entries, and every
+  loaded Python module file outside the project.
+- Every artifact the stored compile wrote under `target/compiled/`, and the `--dag` file. If one is
+  missing or changed, compile runs in full and rewrites it. A compile is stored only when these
+  artifacts still hold the bytes it wrote, so a concurrent compile cannot leave another compile's
+  artifacts behind a reused result.
+
+These project paths are not inputs: `target/`, `logs/`, `venv/`, `.git/` and `__pycache__/`
+anywhere, and these folders at the project root: `.cache`, `.fensu`, `.hg`, `.idea`,
+`.mypy_cache`, `.nox`, `.pytest_cache`, `.ruff_cache`, `.sqlbuild`, `.svn`, `.tox`, `.venv`, and
+`.vscode`. Other hidden folders are inputs, because custom rules can read them. DuckDB database
+files count only by presence.
+
+Reuse assumes macros, hooks, providers, adapters, and custom rules are deterministic: the same inputs
+must produce the same output. A compile whose templates read `run.id` is never reused.
+
+Compile always runs in full with `--no-cache`, `--manifest`, `--debug`, or a profiling flag, when
+`SQLBUILD_DISABLE_COMPILE_CACHE=1` or `SQLBUILD_DISABLE_COMPILE_REUSE=1` is set, when the project has
+no `sqlbuild_project.toml` or `sqlbuild_project.yml`, when the target sets `compile_cache = false`,
+and when a provider's settings use sources SQLBuild cannot list exactly, such as a custom
+`settings_customise_sources` or command-line parsing. The reason is logged at debug level.
+
+SQLBuild keeps one stored result per project and target under
+`target/cache/compiler/project-reuse-v1/`. It is replaced atomically; a damaged or unreadable entry is
+ignored and compile runs in full. Storing a result normally takes a fraction of a second after the
+report is ready. When it has to hash a lot of data or the project has very many files, compile
+prints `Recording compile for reuse...` and a completion line to stderr before the report.
+
+Known limits:
+
+- Python packages outside the project are checked by the stamps of the module files the compile
+  loaded. A newly added submodule or package data file that the compile reads without importing it
+  first is not tracked.
+- On network filesystems that cache file attributes, or when clocks differ between machines,
+  timestamps may not reflect a recent edit.
+- On Windows, a file's change time is its creation time. An in-place rewrite that keeps the same
+  size and restores the previous modification time, such as `touch -r`, `rsync -t`, or
+  `robocopy /COPY:T`, is not detected unless the file changed within two seconds of the stored
+  compile. The same applies to filesystems without a real change time, such as some FUSE, SMB, and
+  exFAT mounts.
+- In these cases, or whenever a result looks stale, run `sqb compile --no-cache` or set
+  `SQLBUILD_DISABLE_COMPILE_REUSE=1`.
+
+A replayed JSON report is byte-identical to the previous one except for `compile_timings`, which
+holds `project_reuse_hits`, `project_reuse_misses`, `project_reuse_bypasses`,
+`project_reuse_check_ms`, and `total_ms`. Full compiles report the same reuse counters alongside
+the phase timings. In text mode, a replayed compile does not print the per-phase progress lines.
+Lifecycle sinks receive only the invocation events for a replayed compile; per-phase `operation`
+events are not emitted because no compile phase runs.
+
 ## Static analysis
 
 When SQL analysis is enabled (default), compile performs static analysis on your models without connecting to the warehouse:

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

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import os
 import time
 from dataclasses import replace
 from functools import partial
@@ -28,6 +29,7 @@
     CompileAnalysis,
     CompileWriteResult,
 )
+from sqlbuild.cli.compile_reuse.constants import COMPILE_ARTIFACT_WRITES
 from sqlbuild.cli.entry.models import CompileProfileFlags
 from sqlbuild.cli.output.models import (
     WrittenTarget,
@@ -319,13 +321,14 @@ def write_compile_dag_artifact(
         dag_path=dag_path,
     )
     resolved_dag_path.parent.mkdir(parents=True, exist_ok=True)
-    resolved_dag_path.write_text(
-        build_dag_json(
-            graph=analysis.graph,
-            project_name=analysis.discovered_inputs.project_config.name,
-            python_graph=python_graph,
-        ),
-        encoding="utf-8",
+    dag_json: str = build_dag_json(
+        graph=analysis.graph,
+        project_name=analysis.discovered_inputs.project_config.name,
+        python_graph=python_graph,
+    )
+    _ = resolved_dag_path.write_text(dag_json, encoding="utf-8")
+    COMPILE_ARTIFACT_WRITES.written(
+        path=resolved_dag_path, contents=dag_json.replace("\n", os.linesep).encode("utf-8")
     )
     _ = complete_compile_phase(
         status=status, message=f"Wrote DAG artifact. ({time.monotonic() - dag_start:.2f}s)"

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

@@ -27,6 +27,7 @@
     SqlTestArtifactCacheRecord,
     SqlTestArtifactIdentityContext,
 )
+from sqlbuild.cli.compile_reuse.constants import COMPILE_ARTIFACT_WRITES
 from sqlbuild.cli.output.models import (
     WrittenTarget,
 )
@@ -235,6 +236,7 @@ def _move_staged_tree(*, staged_dir: Path, path: Path) -> bool:
             os.rename(staged_dir, path)
         except OSError:
             return False
+    COMPILE_ARTIFACT_WRITES.moved_tree(source=staged_dir, destination=path)
     return True
 
 
@@ -250,6 +252,7 @@ def _staged_files(*, staged_dir: Path) -> list[tuple[Path, str]]:
 
 def _publish_staged_file(*, source: Path, path: Path, check_existing: bool) -> None:
     with record_compile_timing("physical_write_ms"):
+        COMPILE_ARTIFACT_WRITES.moved(source=source, destination=path)
         if check_existing and path.is_file():
             contents: bytes = source.read_bytes()
             existing: bytes = path.read_bytes()
@@ -542,6 +545,11 @@ def _partition_cached_static_tests(
                     identity=artifact_identity,
                 )
                 if cached_path is not None:
+                    COMPILE_ARTIFACT_WRITES.kept(
+                        path=cached_path,
+                        size=cached_record.size,
+                        mtime_ns=cached_record.mtime_ns,
+                    )
                     managed_paths.add(cached_path)
                     current_records[record_key] = cached_record
                     continue
@@ -664,6 +672,10 @@ def _write_sql(*, path: Path, sql: str, check_existing: bool = True) -> None:
 
 def _write_text_if_changed(*, path: Path, contents: str, check_existing: bool = True) -> None:
     with record_compile_timing("physical_write_ms"):
+        COMPILE_ARTIFACT_WRITES.written(
+            path=path,
+            contents=contents.replace(_POSIX_LINE_SEPARATOR, os.linesep).encode("utf-8"),
+        )
         if check_existing and path.is_file() and path.read_text(encoding="utf-8") == contents:
             return
         try:
@@ -675,6 +687,7 @@ def _write_text_if_changed(*, path: Path, contents: str, check_existing: bool =
 
 def _write_bytes_if_changed(*, path: Path, contents: bytes, check_existing: bool = True) -> None:
     with record_compile_timing("physical_write_ms"):
+        COMPILE_ARTIFACT_WRITES.written(path=path, contents=contents)
         if check_existing:
             existing: bytes | None = _read_existing_file(path=path)
             if existing == contents:

src/sqlbuild/cli/commands/_helpers/entry/dispatch.py

@@ -501,6 +501,7 @@ def _dispatch_compile_or_contract(
                     skip_contracts=args.profile_skip_contracts,
                     skip_write=args.profile_skip_write,
                 ),
+                debug=args.debug,
             )
         )
     return None

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

@@ -2,12 +2,12 @@
 
 from __future__ import annotations
 
-import sys
 import time
 from dataclasses import replace
 from pathlib import Path
 
 from sqlbuild.cli.commands._helpers.compile.collection import paused_cyclic_collection
+from sqlbuild.cli.commands._helpers.compile.dag import resolve_compile_dag_path
 from sqlbuild.cli.commands._helpers.compile.output import (
     format_compile_json,
     format_compile_text,
@@ -25,9 +25,17 @@
 from sqlbuild.cli.commands.types import CompileLineageMode
 from sqlbuild.cli.compile.models import (
     CompileAnalysis,
+    CompileCommandResult,
     CompileWriteResult,
 )
+from sqlbuild.cli.compile_reuse.classes.recorded_compile_output import RecordedCompileOutput
+from sqlbuild.cli.compile_reuse.constants import COMPILE_ARTIFACT_WRITES
+from sqlbuild.cli.compile_reuse.main._compile_reuse_timings import compile_reuse_timings
+from sqlbuild.cli.compile_reuse.main._write_reusable_compile import write_reusable_compile
+from sqlbuild.cli.compile_reuse.models import CompileReuseAttempt
+from sqlbuild.cli.compile_reuse.types import CompileReuseOutcome
 from sqlbuild.cli.entry.models import CompileCommandRequest
+from sqlbuild.compiler.compile.constants import COMPILE_INPUT_READS
 from sqlbuild.compiler.compile.models import CompileAnalysisSelection, CompilerDiagnostic
 from sqlbuild.compiler.compile.types import DiagnosticPhase
 from sqlbuild.compiler.profiling.main.collect import collect_compile_timings
@@ -37,13 +45,25 @@
 from sqlbuild.rule_engine.main.render_skipped_rules import format_skipped_type_proof_rules
 
 
-def run_compile(request: CompileCommandRequest) -> int:
+def run_compile(
+    *, request: CompileCommandRequest, reuse_attempt: CompileReuseAttempt | None = None
+) -> int:
     """Execute the compile command."""
 
-    total_start: float = time.monotonic()
+    total_start: float = time.monotonic() if reuse_attempt is None else reuse_attempt.started
     effective_request: CompileCommandRequest = (
         request if request.project_dir is not None else replace(request, project_dir=Path.cwd())
     )
+    attempt: CompileReuseAttempt = (
+        CompileReuseAttempt(
+            outcome=CompileReuseOutcome.BYPASS,
+            project_dir=Path(effective_request.project_dir or Path.cwd()),
+            started=total_start,
+        )
+        if reuse_attempt is None
+        else reuse_attempt
+    )
+    output: RecordedCompileOutput = RecordedCompileOutput()
     status: TransientStatusReporter | None = start_compile_status(
         json_output=request.json_output,
         no_color=request.no_color,
@@ -52,20 +72,40 @@ def run_compile(request: CompileCommandRequest) -> int:
         with (
             paused_cyclic_collection(),
             collect_compile_timings() as detailed_timings,
+            COMPILE_INPUT_READS.recording() as input_reads,
+            COMPILE_ARTIFACT_WRITES.recording() as artifact_writes,
             PreparedCompileArtifacts(
                 enabled=not request.profile_flags.skip_write
             ) as prepared_artifacts,
         ):
-            return _run_compile_with_status(
+            result: CompileCommandResult = _run_compile_with_status(
                 request=effective_request,
                 total_start=total_start,
                 status=status,
                 detailed_timings=detailed_timings,
                 prepared_artifacts=prepared_artifacts,
+                output=output,
+                reuse_timings_ms=compile_reuse_timings(attempt=attempt),
             )
     finally:
         if status is not None:
             status.close()
+    try:
+        with paused_cyclic_collection():
+            write_reusable_compile(
+                attempt=attempt,
+                output=output,
+                exit_code=result.exit_code,
+                input_reads=input_reads,
+                artifact_writes=artifact_writes,
+                compile_cache_enabled=result.compile_cache_enabled,
+                artifacts_written=result.artifacts_written,
+                dag_artifact_path=result.dag_artifact_path,
+                json_output=request.json_output,
+            )
+    finally:
+        output.print_stdout()
+    return result.exit_code
 
 
 def _run_compile_with_status(
@@ -75,7 +115,9 @@ def _run_compile_with_status(
     status: TransientStatusReporter | None,
     detailed_timings: CompileTimingCollector,
     prepared_artifacts: PreparedCompileArtifacts,
-) -> int:
+    output: RecordedCompileOutput,
+    reuse_timings_ms: dict[str, int],
+) -> CompileCommandResult:
     """Execute compile after the optional interactive status reporter is initialized."""
 
     project_dir: Path = request.project_dir if request.project_dir is not None else Path.cwd()
@@ -132,6 +174,7 @@ def _run_compile_with_status(
         "rule_cache_misses": analysis.rule_cache_misses,
         "write_ms": write_result.write_ms,
         **detailed_timings.as_milliseconds(),
+        **reuse_timings_ms,
         "total_ms": elapsed_ms(total_start),
     }
     withheld_test_diagnostics: tuple[CompilerDiagnostic, ...] = (
@@ -154,7 +197,14 @@ def _run_compile_with_status(
         codes=analysis.skipped_type_proof_rules
     )
     if skipped_rules_note is not None:
-        print(f"note: {skipped_rules_note}", file=sys.stderr)
+        output.print_stderr_line(f"note: {skipped_rules_note}")
+    result: CompileCommandResult = _command_result(
+        request=request,
+        project_dir=project_dir,
+        exit_code=exit_code,
+        rules_failed=rules_failed,
+        analysis=analysis,
+    )
 
     if json_output:
         notice: str | None = semantic_coverage_notice(
@@ -163,8 +213,8 @@ def _run_compile_with_status(
             sql_validation_enabled=not request.no_sql_validation,
         )
         if notice:
-            print(notice, file=sys.stderr)
-        print(
+            output.print_stderr_line(notice)
+        output.record_stdout(
             format_compile_json(
                 graph=analysis.graph,
                 written=write_result.written,
@@ -177,9 +227,9 @@ def _run_compile_with_status(
                 sql_validation_enabled=not request.no_sql_validation,
             )
         )
-        return exit_code
+        return result
 
-    print(
+    output.record_stdout(
         format_compile_text(
             graph=analysis.graph,
             written=write_result.written,
@@ -192,4 +242,24 @@ def _run_compile_with_status(
             sql_validation_enabled=not request.no_sql_validation,
         )
     )
-    return exit_code
+    return result
+
+
+def _command_result(
+    *,
+    request: CompileCommandRequest,
+    project_dir: Path,
+    exit_code: int,
+    rules_failed: bool,
+    analysis: CompileAnalysis,
+) -> CompileCommandResult:
+    return CompileCommandResult(
+        exit_code=exit_code,
+        compile_cache_enabled=analysis.graph.project.compile_cache_dir is not None,
+        artifacts_written=not (rules_failed or request.profile_flags.skip_write),
+        dag_artifact_path=(
+            None
+            if rules_failed or request.dag_path is None
+            else resolve_compile_dag_path(project_dir=project_dir, dag_path=request.dag_path)
+        ),
+    )