Refine and document project-index linting

Signed-off-by: Andreas Fredhøi <andreas.fredhoi@fresio.no>

Author: fresioAS

docs/guides/configuration.md

@@ -329,6 +329,16 @@ By default, the SQLMesh cache is stored in a `.cache` directory within your proj
 
 The cache directory is automatically created if it doesn't exist. You can clear the cache using the `sqlmesh clean` command.
 
+#### Project index
+
+The `--use-project-index` option on supported commands maintains a persistent model dependency index in the cache directory. Each project writes a file named `<project>_<hash>_model_index.json`.
+
+A full project load with the option enabled creates or refreshes the index. SQLMesh invalidates it when relevant configuration, gateway, macro, audit, or signal metadata changes, or when the set of model files changes. If the index is missing, invalid, or stale, SQLMesh safely falls back to a full project load and rebuilds it.
+
+For operations targeting selected models, the index allows SQLMesh to load only those models and their upstream dependencies.
+
+In multi-repository projects, dependencies that cross project boundaries may not be represented by an individual project's index. SQLMesh detects incomplete scoped loads and falls back to loading the full configured project set.
+
 ### Table/view storage locations
 
 SQLMesh creates schemas, physical tables, and views in the data warehouse/engine. Learn more about why and how SQLMesh creates schema in the ["Why does SQLMesh create schemas?" FAQ](../faq/faq.md#schema-question).

docs/reference/cli.md

@@ -650,9 +650,12 @@ Usage: sqlmesh lint [OPTIONS]
 
 Options:
   --model TEXT           A model to lint. Multiple models can be linted.  If no models are specified, every model will be linted.
+  --use-project-index    Use the persistent project index. With --model, only the selected models and their upstream dependencies
+                         are loaded, resolved, and validated, so errors in unrelated models are not reported. Without --model,
+                         every model is still loaded and linted.
   --local                Lint using only locally loaded project files without loading state. In multi-repository setups, or when
                          linting only a subset of projects, this may cause additional linting errors because SQLMesh will not resolve
                          references or schemas from models that exist only in remote state.
   --help                 Show this message and exit.
 
-```
+```

sqlmesh/core/context.py

@@ -3419,32 +3419,30 @@ def lint_models(
         raise_on_error: bool = True,
         use_project_index: bool = False,
     ) -> t.List[AnnotatedRuleViolation]:
-        input_models = list(models) if models is not None else []
+        models = list(models) if models is not None else []
 
         if not self._loaded:
-            if input_models and use_project_index:
-                target_fqns = {
+            target_fqns = (
+                {
                     normalize_model_name(
                         model,
                         default_catalog=self.default_catalog,
                         dialect=self.default_dialect,
                     )
                     if isinstance(model, str)
                     else model.fqn
-                    for model in input_models
+                    for model in models
                 }
-                self.load(
-                    model_fqns=target_fqns,
-                    use_project_index=True,
-                )
-            else:
-                self.load(use_project_index=use_project_index)
+                if models and use_project_index
+                else None
+            )
+            self.load(model_fqns=target_fqns, use_project_index=use_project_index)
 
         found_error = False
 
         model_list = (
-            list(self.get_model(model, raise_if_missing=True) for model in input_models)
-            if input_models
+            list(self.get_model(model, raise_if_missing=True) for model in models)
+            if models
             else self.models.values()
         )
         all_violations = []

sqlmesh/core/loader.py

@@ -7,6 +7,7 @@
 import linecache
 import os
 import re
+import tempfile
 import typing as t
 from collections import Counter, defaultdict
 from dataclasses import dataclass
@@ -605,9 +606,9 @@ def _model_index_path(self) -> Path:
             / f"{self.config.project or 'default'}_{project_path_hash}_model_index.json"
         )
 
-    def _model_index_id(self) -> t.List[t.Optional[t.Union[str, float]]]:
+    def _model_index_id(self) -> t.List[t.Optional[t.Union[str, int, float]]]:
         return [
-            str(self.MODEL_INDEX_VERSION),
+            self.MODEL_INDEX_VERSION,
             self.config.fingerprint,
             self.context.default_catalog,
             self.context.gateway or self.config.default_gateway_name,
@@ -640,20 +641,35 @@ def _selected_model_paths(
         except (OSError, ValueError):
             return None
 
+        if not isinstance(index, dict):
+            return None
+
         if index.get("index_id") != self._model_index_id():
             return None
 
         current_paths = self._model_paths()
-        indexed_files = index.get("files", {})
+        indexed_files = index.get("files")
+        if not isinstance(indexed_files, dict) or not all(
+            isinstance(relative_path, str) and isinstance(file_models, dict)
+            for relative_path, file_models in indexed_files.items()
+        ):
+            return None
+
         indexed_paths = {self.config_path / relative_path for relative_path in indexed_files}
         if current_paths != indexed_paths:
             return None
 
         model_to_path: t.Dict[str, Path] = {}
         dependencies: t.Dict[str, t.Set[str]] = {}
-        for relative_path, file_info in indexed_files.items():
+        for relative_path, file_models in indexed_files.items():
             path = self.config_path / relative_path
-            for fqn, depends_on in file_info.get("models", {}).items():
+            for fqn, depends_on in file_models.items():
+                if (
+                    not isinstance(fqn, str)
+                    or not isinstance(depends_on, list)
+                    or not all(isinstance(dependency, str) for dependency in depends_on)
+                ):
+                    return None
                 model_to_path[fqn] = path
                 dependencies[fqn] = set(depends_on)
 
@@ -673,25 +689,36 @@ def _write_model_index(self, models: UniqueKeyDict[str, Model]) -> None:
         if not model_paths:
             return
 
-        files: t.Dict[str, t.Dict[str, t.Any]] = {}
-        for path in model_paths:
-            files[str(path.relative_to(self.config_path))] = {
-                "models": {},
-            }
-
+        # Maps each model file to the models it defines and their dependencies.
+        files: t.Dict[str, t.Dict[str, t.List[str]]] = {
+            str(path.relative_to(self.config_path)): {} for path in model_paths
+        }
         for model in models.values():
-            if model._path not in model_paths:
-                continue
-            relative_path = str(t.cast(Path, model._path).relative_to(self.config_path))
-            files[relative_path]["models"][model.fqn] = sorted(model.depends_on)
+            if model._path in model_paths:
+                relative_path = str(t.cast(Path, model._path).relative_to(self.config_path))
+                files[relative_path][model.fqn] = sorted(model.depends_on)
 
         self._model_index_path.parent.mkdir(parents=True, exist_ok=True)
-        temporary_path = self._model_index_path.with_suffix(".tmp")
-        temporary_path.write_text(
-            json.dumps({"index_id": self._model_index_id(), "files": files}, sort_keys=True),
-            encoding="utf-8",
-        )
-        temporary_path.replace(self._model_index_path)
+        temporary_path: t.Optional[Path] = None
+        try:
+            with tempfile.NamedTemporaryFile(
+                mode="w",
+                encoding="utf-8",
+                dir=self._model_index_path.parent,
+                prefix=f".{self._model_index_path.name}.",
+                suffix=".tmp",
+                delete=False,
+            ) as temporary_file:
+                json.dump(
+                    {"index_id": self._model_index_id(), "files": files},
+                    temporary_file,
+                    sort_keys=True,
+                )
+                temporary_path = Path(temporary_file.name)
+            temporary_path.replace(self._model_index_path)
+        finally:
+            if temporary_path is not None:
+                temporary_path.unlink(missing_ok=True)
 
     def _load_sql_models(
         self,

tests/core/test_context.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import pathlib
 import typing as t
@@ -1403,13 +1404,13 @@ def get_sushi_fingerprints(context: Context):
 def test_physical_schema_mapping(tmp_path: pathlib.Path) -> None:
     create_temp_file(
         tmp_path,
-        pathlib.Path("models", "a.sql"),
+        pathlib.Path(pathlib.Path("models"), "a.sql"),
         "MODEL(name foo_staging.model_a); SELECT 1;",
     )
 
     create_temp_file(
         tmp_path,
-        pathlib.Path("models", "b.sql"),
+        pathlib.Path(pathlib.Path("models"), "b.sql"),
         "MODEL(name testone.model_b); SELECT 1;",
     )
 
@@ -2951,7 +2952,7 @@ def create_context() -> Context:
 
     create_temp_file(
         tmp_path,
-        pathlib.Path(pathlib.Path("models"), "a.sql"),
+        pathlib.Path("models", "a.sql"),
         "MODEL(name a); SELECT 1 AS col FROM raw.unregistered_source;",
     )
     create_temp_file(tmp_path, pathlib.Path("models", "b.sql"), "MODEL(name b); SELECT col FROM a;")
@@ -2990,7 +2991,7 @@ def create_context() -> Context:
     # dependency outside this set, Context.load safely retries with a full load.
     create_temp_file(
         tmp_path,
-        pathlib.Path(pathlib.Path("models"), "b.sql"),
+        pathlib.Path("models", "b.sql"),
         "MODEL(name b); SELECT col + 1 AS col FROM a;",
     )
     ctx = create_context()
@@ -3052,6 +3053,49 @@ def create_context() -> Context:
     assert set(schemas_mock.call_args.kwargs["models"]) == set(ctx.models)
 
 
+@pytest.mark.parametrize("invalid_files_shape", ["file_list", "model_list"])
+def test_invalid_model_index_falls_back_to_full_load(
+    tmp_path: pathlib.Path, invalid_files_shape: str
+) -> None:
+    create_temp_file(
+        tmp_path,
+        pathlib.Path("models", "a.sql"),
+        "MODEL(name a); SELECT 1 AS col;",
+    )
+    create_temp_file(
+        tmp_path,
+        pathlib.Path("models", "b.sql"),
+        "MODEL(name b); SELECT col FROM a;",
+    )
+    config = Config(model_defaults=ModelDefaultsConfig(dialect="duckdb"))
+
+    indexed_context = Context(config=config, paths=tmp_path, load=False)
+    indexed_context.load(use_project_index=True)
+    indexed_loader = t.cast(SqlMeshLoader, indexed_context._loaders[0])
+    index = json.loads(indexed_loader._model_index_path.read_text(encoding="utf-8"))
+    if invalid_files_shape == "file_list":
+        index["files"] = list(index["files"])
+    else:
+        relative_path = next(iter(index["files"]))
+        index["files"][relative_path] = []
+    indexed_loader._model_index_path.write_text(json.dumps(index), encoding="utf-8")
+
+    context = Context(config=config, paths=tmp_path, load=False)
+    loader = t.cast(SqlMeshLoader, context._loaders[0])
+    with patch.object(
+        loader,
+        "_load_sql_models",
+        wraps=loader._load_sql_models,
+    ) as load_sql_models_mock:
+        assert context.lint_models(["b"], use_project_index=True) == []
+
+    assert load_sql_models_mock.call_args.kwargs["selected_paths"] is None
+    assert set(context.models) == {
+        context.get_model("a", raise_if_missing=True).fqn,
+        context.get_model("b", raise_if_missing=True).fqn,
+    }
+
+
 def test_plan_selector_expression_no_match(sushi_context: Context) -> None:
     with pytest.raises(
         PlanError,