From 20035dabb7b037765cb46d946ccfdafed021b4b0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andreas=20Fredh=C3=B8i?= Date: Mon, 20 Jul 2026 23:34:33 +0200 Subject: [PATCH 1/2] Perf: add opt-in project-index loading for lint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add --use-project-index to targeted and full lint. The default path preserves the existing full context load. With both --use-project-index and --model, SQLMesh uses a persistent model-to-file dependency index to load, resolve, and validate only the selected models and their transitive upstream dependencies. A full indexed lint still loads and lints every model, but creates or refreshes the index for later targeted commands. Missing or expanded dependencies safely retry with a full indexed load. Co-Authored-By: Claude Fable 5 Signed-off-by: Andreas Fredhøi --- sqlmesh/cli/main.py | 20 ++++- sqlmesh/core/context.py | 139 ++++++++++++++++++++++++------- sqlmesh/core/loader.py | 164 +++++++++++++++++++++++++++++++++++-- sqlmesh/dbt/loader.py | 17 +++- tests/cli/test_cli.py | 45 ++++++++++ tests/core/test_context.py | 121 ++++++++++++++++++++++++++- 6 files changed, 461 insertions(+), 45 deletions(-) diff --git a/sqlmesh/cli/main.py b/sqlmesh/cli/main.py index f5a3c62cf9..2e7da2a2e8 100644 --- a/sqlmesh/cli/main.py +++ b/sqlmesh/cli/main.py @@ -141,6 +141,10 @@ def cli( if ctx.invoked_subcommand in SKIP_LOAD_COMMANDS: load = False + # Unlike the other commands above, lint can scope its own load for multi-project contexts. + if ctx.invoked_subcommand == "lint": + load = False + configs = load_configs(config, Context.CONFIG_TYPE, paths, dotenv_path=dotenv) log_limit = list(configs.values())[0].log_limit @@ -1208,6 +1212,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 +1229,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) diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py index f30c401a36..97c7fb4c7d 100644 --- a/sqlmesh/core/context.py +++ b/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, @@ -3358,12 +3417,34 @@ 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]: + input_models = list(models) if models is not None else [] + + if not self._loaded: + if input_models and use_project_index: + 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 + } + self.load( + model_fqns=target_fqns, + use_project_index=True, + ) + else: + self.load(use_project_index=use_project_index) + found_error = False model_list = ( - list(self.get_model(model, raise_if_missing=True) for model in models) - if models + list(self.get_model(model, raise_if_missing=True) for model in input_models) + if input_models else self.models.values() ) all_violations = [] diff --git a/sqlmesh/core/loader.py b/sqlmesh/core/loader.py index cb951b4f9e..9bfa4988f8 100644 --- a/sqlmesh/core/loader.py +++ b/sqlmesh/core/loader.py @@ -3,6 +3,7 @@ import abc import glob import itertools +import json import linecache import os import re @@ -38,6 +39,7 @@ from sqlmesh.core.test import ModelTestMetadata from sqlmesh.utils import UniqueKeyDict, sys_path from sqlmesh.utils.errors import ConfigError +from sqlmesh.utils.hashing import crc32 from sqlmesh.utils.jinja import JinjaMacroRegistry, MacroExtractor from sqlmesh.utils.metaprogramming import import_python_file from sqlmesh.utils.pydantic import validation_error_message @@ -65,6 +67,7 @@ class LoadedProject: environment_statements: t.List[EnvironmentStatements] user_rules: RuleSet model_test_metadata: t.List[ModelTestMetadata] + indexed_model_fqns: t.Optional[t.Set[str]] = None class CacheBase(abc.ABC): @@ -188,10 +191,19 @@ def __init__(self, context: GenericContext, path: Path) -> None: } _init_model_defaults(self.config_essentials, self.context.selected_gateway) - def load(self) -> LoadedProject: + def load( + self, + model_fqns: t.Optional[t.Set[str]] = None, + use_project_index: bool = False, + ) -> LoadedProject: """ Loads all macros and models in the context's path. + Args: + model_fqns: Optional model names to load. Loaders that support partial loading + also include these models' transitive upstream dependencies. + use_project_index: Whether to use and maintain the persistent project model index. + Returns: A loaded project object. """ @@ -228,12 +240,14 @@ def load(self) -> LoadedProject: else: standalone_audits[name] = audit - models = self._load_models( + models, indexed_model_fqns = self._load_models( macros, jinja_macros, self.context.selected_gateway, audits, signals, + model_fqns, + use_project_index, ) metrics = self._load_metrics() @@ -258,6 +272,7 @@ def load(self) -> LoadedProject: environment_statements=environment_statements, user_rules=user_rules, model_test_metadata=model_test_metadata, + indexed_model_fqns=indexed_model_fqns, ) return project @@ -286,7 +301,9 @@ def _load_models( gateway: t.Optional[str], audits: UniqueKeyDict[str, ModelAudit], signals: UniqueKeyDict[str, signal], - ) -> UniqueKeyDict[str, Model]: + model_fqns: t.Optional[t.Set[str]] = None, + use_project_index: bool = False, + ) -> t.Tuple[UniqueKeyDict[str, Model], t.Optional[t.Set[str]]]: """Loads all models.""" @abc.abstractmethod @@ -481,6 +498,10 @@ def _failed_to_load_model_error(self, path: Path, error: t.Union[str, Exception] class SqlMeshLoader(Loader): """Loads macros and models for a context using the SQLMesh file formats""" + MODEL_INDEX_VERSION = 1 + _signals_max_mtime: t.Optional[float] + _audits_max_mtime: t.Optional[float] + def _load_scripts(self) -> t.Tuple[MacroRegistry, JinjaMacroRegistry]: """Loads all user defined macros.""" # Store a copy of the macro registry @@ -533,23 +554,144 @@ def _load_models( gateway: t.Optional[str], audits: UniqueKeyDict[str, ModelAudit], signals: UniqueKeyDict[str, signal], - ) -> UniqueKeyDict[str, Model]: + model_fqns: t.Optional[t.Set[str]] = None, + use_project_index: bool = False, + ) -> t.Tuple[UniqueKeyDict[str, Model], t.Optional[t.Set[str]]]: """ Loads all of the models within the model directory with their associated audits into a Dict and creates the dag """ cache = SqlMeshLoader._Cache(self, self.config_path) - - sql_models = self._load_sql_models(macros, jinja_macros, audits, signals, cache, gateway) + selected_model_index = ( + self._selected_model_paths(model_fqns) if use_project_index and model_fqns else None + ) + selected_paths, indexed_model_fqns = selected_model_index or (None, None) + + sql_models = self._load_sql_models( + macros, + jinja_macros, + audits, + signals, + cache, + gateway, + selected_paths=selected_paths, + ) external_models = self._load_external_models(audits, cache, gateway) - python_models = self._load_python_models(macros, jinja_macros, audits, signals) + python_models = self._load_python_models( + macros, + jinja_macros, + audits, + signals, + selected_paths=selected_paths, + ) all_model_names = list(sql_models) + list(external_models) + list(python_models) duplicates = [name for name, count in Counter(all_model_names).items() if count > 1] if duplicates: raise ConfigError(f"Duplicate model name(s) found: {', '.join(duplicates)}.") - return UniqueKeyDict("models", **sql_models, **external_models, **python_models) + models: UniqueKeyDict[str, Model] = UniqueKeyDict( + "models", **sql_models, **external_models, **python_models + ) + if use_project_index and selected_paths is None: + self._write_model_index(models) + return models, indexed_model_fqns + + @property + def _model_index_path(self) -> Path: + project_path_hash = crc32([str(self.config_path.resolve())]) + return ( + self.context.cache_dir + / 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]]]: + return [ + str(self.MODEL_INDEX_VERSION), + self.config.fingerprint, + self.context.default_catalog, + self.context.gateway or self.config.default_gateway_name, + self._macros_max_mtime, + self._signals_max_mtime, + self._audits_max_mtime, + self._config_mtimes.get(self.config_path), + self._config_mtimes.get(c.SQLMESH_PATH), + ] + + def _model_paths(self) -> t.Set[Path]: + paths: t.Set[Path] = set() + for extension in (".sql", ".py"): + paths.update( + path + for path in self._glob_paths( + self.config_path / c.MODELS, + ignore_patterns=self.config.ignore_patterns, + extension=extension, + ) + if os.path.getsize(path) + ) + return paths + + def _selected_model_paths( + self, model_fqns: t.Set[str] + ) -> t.Optional[t.Tuple[t.Set[Path], t.Set[str]]]: + try: + index = json.loads(self._model_index_path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + + if index.get("index_id") != self._model_index_id(): + return None + + current_paths = self._model_paths() + indexed_files = index.get("files", {}) + 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(): + path = self.config_path / relative_path + for fqn, depends_on in file_info.get("models", {}).items(): + model_to_path[fqn] = path + dependencies[fqn] = set(depends_on) + + selected = {fqn for fqn in model_fqns if fqn in model_to_path} + stack = list(selected) + while stack: + fqn = stack.pop() + for dependency in dependencies.get(fqn, set()): + if dependency in model_to_path and dependency not in selected: + selected.add(dependency) + stack.append(dependency) + + return {model_to_path[fqn] for fqn in selected}, set(model_to_path) + + def _write_model_index(self, models: UniqueKeyDict[str, Model]) -> None: + model_paths = self._model_paths() + 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": {}, + } + + 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) + + 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) def _load_sql_models( self, @@ -559,6 +701,7 @@ def _load_sql_models( signals: UniqueKeyDict[str, signal], cache: CacheBase, gateway: t.Optional[str], + selected_paths: t.Optional[t.Set[Path]] = None, ) -> UniqueKeyDict[str, Model]: """Loads the sql models into a Dict""" models: UniqueKeyDict[str, Model] = UniqueKeyDict("models") @@ -570,6 +713,8 @@ def _load_sql_models( ignore_patterns=self.config.ignore_patterns, extension=".sql", ): + if selected_paths is not None and path not in selected_paths: + continue if not os.path.getsize(path): continue @@ -641,6 +786,7 @@ def _load_python_models( jinja_macros: JinjaMacroRegistry, audits: UniqueKeyDict[str, ModelAudit], signals: UniqueKeyDict[str, signal], + selected_paths: t.Optional[t.Set[Path]] = None, ) -> UniqueKeyDict[str, Model]: """Loads the python models into a Dict""" models: UniqueKeyDict[str, Model] = UniqueKeyDict("models") @@ -655,6 +801,8 @@ def _load_python_models( ignore_patterns=self.config.ignore_patterns, extension=".py", ): + if selected_paths is not None and path not in selected_paths: + continue if not os.path.getsize(path): continue diff --git a/sqlmesh/dbt/loader.py b/sqlmesh/dbt/loader.py index fb3ecb2c77..ac4764f2af 100644 --- a/sqlmesh/dbt/loader.py +++ b/sqlmesh/dbt/loader.py @@ -133,9 +133,16 @@ def __init__( self._profiles_dir = profiles_dir super().__init__(context, path) - def load(self) -> LoadedProject: + def load( + self, + model_fqns: t.Optional[t.Set[str]] = None, + use_project_index: bool = False, + ) -> LoadedProject: self._projects = [] - return super().load() + return super().load( + model_fqns=model_fqns, + use_project_index=use_project_index, + ) def _load_scripts(self) -> t.Tuple[MacroRegistry, JinjaMacroRegistry]: macro_files = list(Path(self.config_path, "macros").glob("**/*.sql")) @@ -156,7 +163,9 @@ def _load_models( gateway: t.Optional[str], audits: UniqueKeyDict[str, ModelAudit], signals: UniqueKeyDict[str, signal], - ) -> UniqueKeyDict[str, Model]: + model_fqns: t.Optional[t.Set[str]] = None, + use_project_index: bool = False, + ) -> t.Tuple[UniqueKeyDict[str, Model], t.Optional[t.Set[str]]]: models: UniqueKeyDict[str, Model] = UniqueKeyDict("models") def _to_sqlmesh(config: BMC, context: DbtContext) -> Model: @@ -200,7 +209,7 @@ def _to_sqlmesh(config: BMC, context: DbtContext) -> Model: models.update(self._load_external_models(audits, cache)) - return models + return models, None def _load_audits( self, macros: MacroRegistry, jinja_macros: JinjaMacroRegistry diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py index 092def8e0c..7d0c925ed3 100644 --- a/tests/cli/test_cli.py +++ b/tests/cli/test_cli.py @@ -1314,6 +1314,51 @@ def test_lint(runner, tmp_path): assert result.exit_code == 1 +def test_lint_no_models(runner, tmp_path): + with open(tmp_path / "config.yaml", "w", encoding="utf-8") as f: + f.write("model_defaults:\n dialect: duckdb\n") + + result = runner.invoke(cli, ["--paths", tmp_path, "lint"]) + assert result.exit_code == 1 + assert "doesn't seem to have any models" in result.output + + +def test_lint_model_scopes_validation_with_multiple_projects(runner, tmp_path): + project_a = tmp_path / "project_a" + project_b = tmp_path / "project_b" + for project in (project_a, project_b): + (project / "models").mkdir(parents=True) + with open(project / "config.yaml", "w", encoding="utf-8") as f: + f.write( + f"project: {project.name}\n" + "model_defaults:\n" + " dialect: duckdb\n" + "linter:\n" + " enabled: true\n" + ) + + with open(project_a / "models" / "selected.sql", "w", encoding="utf-8") as f: + f.write("MODEL(name selected); SELECT 1 AS col;") + with open(project_b / "models" / "unrelated.sql", "w", encoding="utf-8") as f: + f.write("MODEL(name unrelated, kind FULL, partitioned_by (col, col)); SELECT 1 AS col;") + + result = runner.invoke( + cli, + [ + "--paths", + project_a, + "--paths", + project_b, + "lint", + "--use-project-index", + "--model", + "selected", + ], + ) + + assert result.exit_code == 0, result.output + + def test_state_export(runner: CliRunner, tmp_path: Path) -> None: create_example_project(tmp_path) diff --git a/tests/core/test_context.py b/tests/core/test_context.py index 74ce0d7916..aaf982e93a 100644 --- a/tests/core/test_context.py +++ b/tests/core/test_context.py @@ -36,9 +36,10 @@ from sqlmesh.core.dialect import parse, schema_ from sqlmesh.core.engine_adapter.duckdb import DuckDBEngineAdapter from sqlmesh.core.environment import Environment, EnvironmentNamingInfo, EnvironmentStatements +from sqlmesh.core.loader import SqlMeshLoader from sqlmesh.core.plan.definition import Plan from sqlmesh.core.macros import MacroEvaluator, RuntimeStage -from sqlmesh.core.model import load_sql_based_model, model, SqlModel, Model +from sqlmesh.core.model import load_sql_based_model, model, SqlModel, Model, update_model_schemas from sqlmesh.core.model.common import ParsableSql from sqlmesh.core.model.cache import OptimizedQueryCache from sqlmesh.core.renderer import render_statements @@ -1402,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(pathlib.Path("models"), "a.sql"), + pathlib.Path("models", "a.sql"), "MODEL(name foo_staging.model_a); SELECT 1;", ) create_temp_file( tmp_path, - pathlib.Path(pathlib.Path("models"), "b.sql"), + pathlib.Path("models", "b.sql"), "MODEL(name testone.model_b); SELECT 1;", ) @@ -2937,6 +2938,120 @@ def model4_entrypoint(context, **kwargs): sushi_context.plan(environment="dev", auto_apply=True, no_prompts=True) +def test_lint_models_scoped_schema_resolution(tmp_path: pathlib.Path) -> None: + def create_context() -> Context: + return Context( + config=Config( + model_defaults=ModelDefaultsConfig(dialect="duckdb"), + linter=LinterConfig(enabled=True, rules=["noselectstar"]), + ), + paths=tmp_path, + load=False, + ) + + create_temp_file( + tmp_path, + pathlib.Path(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;") + create_temp_file(tmp_path, pathlib.Path("models", "c.sql"), "MODEL(name c); SELECT * FROM b;") + create_temp_file(tmp_path, pathlib.Path("models", "d.sql"), "MODEL(name d); SELECT 2 AS col;") + + # Case: Without the opt-in, linting a specific model preserves the full-load behavior. + ctx = create_context() + with patch( + "sqlmesh.core.context.update_model_schemas", wraps=update_model_schemas + ) as schemas_mock: + violations = ctx.lint_models(["b"]) + + assert violations == [] + assert schemas_mock.call_count == 1 + assert set(schemas_mock.call_args.kwargs["models"]) == set(ctx.models) + + # Case: The project-index opt-in scopes schema resolution to the target and its + # upstream dependencies. The initial full file load also creates the index. + ctx = create_context() + with patch( + "sqlmesh.core.context.update_model_schemas", wraps=update_model_schemas + ) as schemas_mock: + violations = ctx.lint_models(["b"], use_project_index=True) + + assert violations == [] + assert schemas_mock.call_count == 1 + updated_models = schemas_mock.call_args.kwargs["models"] + assert set(updated_models) == { + ctx.get_model("a", raise_if_missing=True).fqn, + ctx.get_model("b", raise_if_missing=True).fqn, + } + + # Case: Once a full load has populated the model index, an edited target still + # loads only its file and upstream dependencies. If the edit introduces a new + # dependency outside this set, Context.load safely retries with a full load. + create_temp_file( + tmp_path, + pathlib.Path(pathlib.Path("models"), "b.sql"), + "MODEL(name b); SELECT col + 1 AS col FROM a;", + ) + ctx = create_context() + loader = t.cast(SqlMeshLoader, ctx._loaders[0]) + with patch.object( + loader, + "_load_sql_models", + wraps=loader._load_sql_models, + ) as load_sql_models_mock: + assert ctx.lint_models(["b"], use_project_index=True) == [] + + assert load_sql_models_mock.call_count == 1 + selected_paths = load_sql_models_mock.call_args.kwargs["selected_paths"] + assert {path.name for path in selected_paths} == {"a.sql", "b.sql"} + assert set(ctx.models) == { + ctx.get_model("a", raise_if_missing=True).fqn, + ctx.get_model("b", raise_if_missing=True).fqn, + } + + # Case: A newly introduced dependency that is known to the index triggers a safe + # full reload instead of leaving the context incomplete. + create_temp_file( + tmp_path, + pathlib.Path("models", "b.sql"), + "MODEL(name b); SELECT col FROM d;", + ) + ctx = create_context() + loader = t.cast(SqlMeshLoader, ctx._loaders[0]) + with patch.object( + loader, + "_load_sql_models", + wraps=loader._load_sql_models, + ) as load_sql_models_mock: + assert ctx.lint_models(["b"], use_project_index=True) == [] + + assert load_sql_models_mock.call_count == 2 + assert set(ctx.models) == { + ctx.get_model(model_name, raise_if_missing=True).fqn for model_name in ("a", "b", "c", "d") + } + + # Case: Violations are still detected when linting specific models. + with pytest.raises( + LinterError, match="Linter detected errors in the code. Please fix them before proceeding." + ): + create_context().lint_models(["c"], use_project_index=True) + + # Case: Linting without a model selection triggers a full load and lints every model. + ctx = create_context() + with patch( + "sqlmesh.core.context.update_model_schemas", wraps=update_model_schemas + ) as schemas_mock: + with pytest.raises( + LinterError, + match="Linter detected errors in the code. Please fix them before proceeding.", + ): + ctx.lint_models() + + assert schemas_mock.call_count == 1 + assert set(schemas_mock.call_args.kwargs["models"]) == set(ctx.models) + + def test_plan_selector_expression_no_match(sushi_context: Context) -> None: with pytest.raises( PlanError, From bcbed2f84743781ef4d125b108602d9ab1b7321a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andreas=20Fredh=C3=B8i?= Date: Fri, 24 Jul 2026 10:29:10 +0100 Subject: [PATCH 2/2] Refine and document project-index linting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Andreas Fredhøi --- docs/guides/configuration.md | 10 ++++++ docs/reference/cli.md | 5 ++- sqlmesh/core/context.py | 22 ++++++------ sqlmesh/core/loader.py | 69 +++++++++++++++++++++++++----------- tests/core/test_context.py | 52 ++++++++++++++++++++++++--- 5 files changed, 120 insertions(+), 38 deletions(-) diff --git a/docs/guides/configuration.md b/docs/guides/configuration.md index b80afa1388..41f4b05594 100644 --- a/docs/guides/configuration.md +++ b/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 `__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). diff --git a/docs/reference/cli.md b/docs/reference/cli.md index 9b6e28fe14..82b4161277 100644 --- a/docs/reference/cli.md +++ b/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. -``` \ No newline at end of file +``` diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py index 97c7fb4c7d..48cb723651 100644 --- a/sqlmesh/core/context.py +++ b/sqlmesh/core/context.py @@ -3419,11 +3419,11 @@ 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, @@ -3431,20 +3431,18 @@ def lint_models( ) 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 = [] diff --git a/sqlmesh/core/loader.py b/sqlmesh/core/loader.py index 9bfa4988f8..5a1ce8b17d 100644 --- a/sqlmesh/core/loader.py +++ b/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, diff --git a/tests/core/test_context.py b/tests/core/test_context.py index aaf982e93a..1f43bf2724 100644 --- a/tests/core/test_context.py +++ b/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,