Refine and document project-index linting

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

Author: fresioAS

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

@@ -651,9 +651,9 @@ def _selected_model_paths(
 
         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():
                 model_to_path[fqn] = path
                 dependencies[fqn] = set(depends_on)
 
@@ -673,17 +673,14 @@ 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")

tests/core/test_context.py

@@ -1403,13 +1403,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 +2951,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 +2990,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()