Perf: Render scope upstream snapshots

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

@@ -447,6 +447,9 @@ Options:
                               only they will be expanded as raw queries.
   --dialect TEXT              The SQL dialect to render the query as.
   --no-format                 Disable fancy formatting of the query.
+  --use-project-index         Use the persistent project index to load and
+                              render only the target model and its upstream
+                              dependencies.
   --max-text-width INTEGER    The max number of characters in a segment before
                               creating new lines in pretty mode.
   --leading-comma             Determines whether or not the comma is leading
@@ -650,9 +653,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/cli/main.py

@@ -141,6 +141,10 @@ def cli(
         if ctx.invoked_subcommand in SKIP_LOAD_COMMANDS:
             load = False
 
+    # These commands can scope their own load for multi-project contexts.
+    if ctx.invoked_subcommand in ("lint", "render"):
+        load = False
+
     configs = load_configs(config, Context.CONFIG_TYPE, paths, dotenv_path=dotenv)
     log_limit = list(configs.values())[0].log_limit
 
@@ -280,6 +284,11 @@ def init(
     help="The SQL dialect to render the query as.",
 )
 @click.option("--no-format", is_flag=True, help="Disable fancy formatting of the query.")
+@click.option(
+    "--use-project-index",
+    is_flag=True,
+    help="Use the persistent project index to load and render only the target model and its upstream dependencies.",
+)
 @opt.format_options
 @click.pass_context
 @error_handler
@@ -293,19 +302,20 @@ def render(
     expand: t.Optional[t.Union[bool, t.Iterable[str]]] = None,
     dialect: t.Optional[str] = None,
     no_format: bool = False,
+    use_project_index: bool = False,
     **format_kwargs: t.Any,
 ) -> None:
     """Render a model's query, optionally expanding referenced models."""
-    model = ctx.obj.get_model(model, raise_if_missing=True)
-
     rendered = ctx.obj.render(
         model,
         start=start,
         end=end,
         execution_time=execution_time,
         expand=expand,
+        use_project_index=use_project_index,
     )
 
+    model = ctx.obj.get_model(model, raise_if_missing=True)
     format_config = ctx.obj.config_for_node(model).format
     format_kwargs = {
         **format_config.generator_options,
@@ -1208,6 +1218,11 @@ def environments(obj: Context) -> None:
     multiple=True,
     help="A model to lint. Multiple models can be linted. If no models are specified, every model will be linted.",
 )
+@click.option(
+    "--use-project-index",
+    is_flag=True,
+    help="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.",
+)
 @click.option(
     "--local",
     is_flag=True,
@@ -1220,9 +1235,18 @@ def environments(obj: Context) -> None:
 def lint(
     obj: Context,
     models: t.Iterator[str],
+    use_project_index: bool,
 ) -> None:
     """Run the linter for the target model(s)."""
-    obj.lint_models(models)
+    obj.lint_models(
+        models,
+        use_project_index=use_project_index,
+    )
+
+    if not obj.models:
+        raise click.ClickException(
+            f"`{obj.path}` doesn't seem to have any models... cd into the proper directory or specify the path(s) with -p."
+        )
 
 
 @cli.group(no_args_is_help=True)

sqlmesh/core/context.py

@@ -418,6 +418,7 @@ def __init__(
         self._linters: t.Dict[str, Linter] = {}
         self._loaded: bool = False
         self._load_state: bool = load_state
+        self._uncached_model_names: t.Set[str] = set()
         self._selector_cls = selector or NativeSelector
 
         self.path, self.config = t.cast(t.Tuple[Path, C], next(iter(self.configs.items())))
@@ -641,11 +642,22 @@ def refresh(self) -> None:
         if any(loader.reload_needed() for loader in self._loaders):
             self.load()
 
-    def load(self, update_schemas: bool = True) -> GenericContext[C]:
-        """Load all files in the context's path."""
+    def load(
+        self,
+        update_schemas: bool = True,
+        model_fqns: t.Optional[t.Set[str]] = None,
+        use_project_index: bool = False,
+    ) -> GenericContext[C]:
+        """Load files in the context's path, optionally scoped to specific models."""
         load_start_ts = time.perf_counter()
 
-        loaded_projects = [loader.load() for loader in self._loaders]
+        loaded_projects = [
+            loader.load(
+                model_fqns=model_fqns,
+                use_project_index=use_project_index,
+            )
+            for loader in self._loaders
+        ]
 
         self.dag = DAG()
         self._standalone_audits.clear()
@@ -688,6 +700,27 @@ def load(self, update_schemas: bool = True) -> GenericContext[C]:
                 BUILTIN_RULES.union(project.user_rules), config.linter
             )
 
+        indexed_model_fqns = {
+            fqn for project in loaded_projects for fqn in (project.indexed_model_fqns or set())
+        }
+        if model_fqns and (
+            not model_fqns <= self._models.keys()
+            or any(
+                dependency in indexed_model_fqns and dependency not in self._models
+                for model in self._models.values()
+                for dependency in model.depends_on
+            )
+        ):
+            # A missing or stale index, a new model, or a dependency crossing project
+            # boundaries requires a full load to preserve existing behavior.
+            self.load(
+                update_schemas=False,
+                use_project_index=use_project_index,
+            )
+            if update_schemas:
+                self._update_model_schemas_and_validate(model_fqns)
+            return self
+
         # Load environment statements from state for projects not in current load
         if self._load_state and any(self._projects):
             prod = self.state_reader.get_environment(c.PROD)
@@ -713,34 +746,13 @@ def load(self, update_schemas: bool = True) -> GenericContext[C]:
                         else:
                             local_store[snapshot.name] = snapshot.node  # type: ignore
 
+        self._uncached_model_names = uncached
+
         for model in self._models.values():
             self.dag.add(model.fqn, model.depends_on)
 
         if update_schemas:
-            for fqn in self.dag:
-                model = self._models.get(fqn)  # type: ignore
-
-                if not model or fqn in uncached:
-                    continue
-
-                # make a copy of remote models that depend on local models or in the downstream chain
-                # without this, a SELECT * FROM local will not propogate properly because the downstream
-                # model will get mutated (schema changes) but the object is the same as the remote cache
-                if any(dep in uncached for dep in model.depends_on):
-                    uncached.add(fqn)
-                    self._models.update({fqn: model.copy(update={"mapping_schema": {}})})
-                    continue
-
-            update_model_schemas(
-                self.dag,
-                models=self._models,
-                cache_dir=self.cache_dir,
-            )
-
-            models = self.models.values()
-            for model in models:
-                # The model definition can be validated correctly only after the schema is set.
-                model.validate_definition()
+            self._update_model_schemas_and_validate(model_fqns or None)
 
         duplicates = set(self._models) & set(self._standalone_audits)
         if duplicates:
@@ -767,6 +779,53 @@ def load(self, update_schemas: bool = True) -> GenericContext[C]:
         self._loaded = True
         return self
 
+    def _update_model_schemas_and_validate(self, model_fqns: t.Optional[t.Set[str]] = None) -> None:
+        """Updates the mapping schemas of the given models (all models by default) and validates their definitions.
+
+        Args:
+            model_fqns: If provided, only these models and their transitive upstream
+                dependencies are processed.
+        """
+        if model_fqns is not None:
+            model_fqns = {
+                fqn for target in model_fqns for fqn in (target, *self.dag.upstream(target))
+            }
+
+        uncached = set(self._uncached_model_names)
+
+        for fqn in self.dag:
+            if model_fqns is not None and fqn not in model_fqns:
+                continue
+
+            model = self._models.get(fqn)
+
+            if not model or fqn in uncached:
+                continue
+
+            # make a copy of remote models that depend on local models or in the downstream chain
+            # without this, a SELECT * FROM local will not propogate properly because the downstream
+            # model will get mutated (schema changes) but the object is the same as the remote cache
+            if any(dep in uncached for dep in model.depends_on):
+                uncached.add(fqn)
+                self._models.update({fqn: model.copy(update={"mapping_schema": {}})})
+                continue
+
+        models = self._models
+        if model_fqns is not None:
+            models = UniqueKeyDict(
+                "models", {fqn: model for fqn, model in self._models.items() if fqn in model_fqns}
+            )
+
+        update_model_schemas(
+            self.dag,
+            models=models,
+            cache_dir=self.cache_dir,
+        )
+
+        for model in models.values():
+            # The model definition can be validated correctly only after the schema is set.
+            model.validate_definition()
+
     @python_api_analytics
     def run(
         self,
@@ -1127,6 +1186,7 @@ def render(
         end: t.Optional[TimeLike] = None,
         execution_time: t.Optional[TimeLike] = None,
         expand: t.Union[bool, t.Iterable[str]] = False,
+        use_project_index: bool = False,
         **kwargs: t.Any,
     ) -> exp.Expr:
         """Renders a model's query, expanding macros with provided kwargs, and optionally expanding referenced models.
@@ -1139,12 +1199,20 @@ def render(
             expand: Whether or not to use expand materialized models, defaults to False.
                 If True, all referenced models are expanded as raw queries.
                 If a list, only referenced models are expanded as raw queries.
+            use_project_index: Whether to use the persistent project index to load and
+                render only the target model and its transitive upstream dependencies.
 
         Returns:
             The rendered expression.
         """
         execution_time = execution_time or now()
 
+        if not self._loaded:
+            target_fqns = (
+                {self._node_or_snapshot_to_fqn(model_or_snapshot)} if use_project_index else None
+            )
+            self.load(model_fqns=target_fqns, use_project_index=use_project_index)
+
         model = self.get_model(model_or_snapshot, raise_if_missing=True)
 
         if expand and not isinstance(expand, bool):
@@ -1173,7 +1241,19 @@ def render(
             )
             return next(pandas_to_sql(t.cast(pd.DataFrame, df), model.columns_to_types))
 
-        snapshots = self.snapshots
+        if use_project_index:
+            # Only the target model and its transitive upstream dependencies can be referenced
+            # by the rendered query, so there is no need to create snapshots for the rest.
+            upstream_fqns = {model.fqn, *self.dag.upstream(model.fqn)}
+            upstream_models: UniqueKeyDict[str, Model] = UniqueKeyDict(
+                "models", {fqn: m for fqn, m in self._models.items() if fqn in upstream_fqns}
+            )
+            snapshots = self._snapshots(
+                upstream_models,
+                include_standalone_audits=False,
+            )
+        else:
+            snapshots = self.snapshots
         deployability_index = DeployabilityIndex.create(snapshots.values(), start=start)
 
         return model.render_query_or_raise(
@@ -2927,9 +3007,13 @@ def _get_engine_adapter(self, gateway: t.Optional[str] = None) -> EngineAdapter:
         return self.engine_adapter
 
     def _snapshots(
-        self, models_override: t.Optional[UniqueKeyDict[str, Model]] = None
+        self,
+        models_override: t.Optional[UniqueKeyDict[str, Model]] = None,
+        include_standalone_audits: bool = True,
     ) -> t.Dict[str, Snapshot]:
-        nodes = {**(models_override or self._models), **self._standalone_audits}
+        nodes: t.Dict[str, Node] = dict(models_override or self._models)
+        if include_standalone_audits:
+            nodes.update(self._standalone_audits)
         snapshots = self._nodes_to_snapshots(nodes)
         stored_snapshots = self.state_reader.get_snapshots(snapshots.values())
 
@@ -3358,7 +3442,27 @@ def lint_models(
         self,
         models: t.Optional[t.Iterable[t.Union[str, Model]]] = None,
         raise_on_error: bool = True,
+        use_project_index: bool = False,
     ) -> t.List[AnnotatedRuleViolation]:
+        models = list(models) if models is not None else []
+
+        if not self._loaded:
+            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 models
+                }
+                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 = (