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/guides/linter.md b/docs/guides/linter.md index 0a2e3ea828..22d8ed9ce5 100644 --- a/docs/guides/linter.md +++ b/docs/guides/linter.md @@ -135,6 +135,21 @@ $ sqlmesh lint --local This can make linting faster in repositories where all referenced models are loaded from local files. In multi-repository setups, or when linting only a subset of projects, `--local` may cause additional linting errors because SQLMesh will not resolve references or schemas from models that exist only in remote state. +For faster targeted linting, enable the persistent project index with `--use-project-index`. When +models are selected with `--model`, SQLMesh loads, resolves, and validates only those models and +their transitive upstream dependencies. The same behavior can be enabled by default for the +Python API and CLI with the `linter.use_project_index` configuration option: + +```yaml +linter: + enabled: true + use_project_index: true +``` + +`Context.lint_models` uses this configuration value when `use_project_index` is omitted. Passing +`use_project_index=False` explicitly disables it for that call. If a context was already loaded, +an indexed lint of selected models reloads the context so the requested scope is applied. + ## Applying linting rules 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/docs/reference/configuration.md b/docs/reference/configuration.md index a7788df73d..a1bf400c32 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -60,6 +60,13 @@ The `model_defaults` key is **required** and must contain a value for the `diale See all the keys allowed in `model_defaults` at the [model configuration reference page](./model_configuration.md#model-defaults). +### Linter + +| Option | Description | Type | Required | +|---------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------|---------|----------| +| `linter.enabled` | Whether linting is enabled (Default: `False`) | boolean | N | +| `linter.use_project_index` | Whether to use the persistent project index for linting. Targeted linting loads selected models and their upstream dependencies. (Default: `False`) | boolean | N | + ### Variables The `variables` key can be used to provide values for user-defined variables, accessed using the [`@VAR` macro function](../concepts/macros/sqlmesh_macros.md#global-variables) in SQL model definitions, [`context.var` method](../concepts/models/python_models.md#global-variables) in Python model definitions, and [`evaluator.var` method](../concepts/macros/sqlmesh_macros.md#accessing-global-variable-values) in Python macro functions. diff --git a/sqlmesh/cli/main.py b/sqlmesh/cli/main.py index 5574a892cb..c2dc1e3dbb 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 @@ -1209,6 +1213,12 @@ 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, + default=None, + 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. Can also be enabled with linter.use_project_index.", +) @click.option( "--local", is_flag=True, @@ -1221,9 +1231,18 @@ def environments(obj: Context) -> None: def lint( obj: Context, models: t.Iterator[str], + use_project_index: t.Optional[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/config/linter.py b/sqlmesh/core/config/linter.py index 11d700c540..0c072e1977 100644 --- a/sqlmesh/core/config/linter.py +++ b/sqlmesh/core/config/linter.py @@ -16,6 +16,8 @@ class LinterConfig(BaseConfig): Args: enabled: Flag indicating whether the linter should run + use_project_index: Whether to use the persistent project index when linting. + rules: A list of error rules to be applied on model warn_rules: A list of rules to be applied on models but produce warnings instead of raising errors. ignored_rules: A list of rules to be excluded/ignored @@ -23,6 +25,7 @@ class LinterConfig(BaseConfig): """ enabled: bool = False + use_project_index: bool = False rules: t.Set[str] = set() warn_rules: t.Set[str] = set() diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py index c3abff1d94..e335253016 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,31 @@ 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. + + Args: + update_schemas: Whether to update model schemas and validate model definitions. + model_fqns: If provided with ``use_project_index=True``, only the selected models + and their transitive upstream dependencies are loaded. + use_project_index: Whether to use and maintain the persistent project model index. + When ``model_fqns`` is not provided, all models are loaded and the index is + refreshed for future scoped loads. + """ 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 +709,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 +755,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 +788,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, @@ -3435,7 +3503,42 @@ def lint_models( self, models: t.Optional[t.Iterable[t.Union[str, Model]]] = None, raise_on_error: bool = True, + use_project_index: t.Optional[bool] = None, ) -> t.List[AnnotatedRuleViolation]: + """Lint the selected models. + + Args: + models: Models to lint. If omitted, all loaded models are linted. + raise_on_error: Whether to raise when an error-level violation is found. + use_project_index: Whether to use the persistent project index. If omitted, the + value of ``linter.use_project_index`` is used. Indexed linting of selected + models reloads an already-loaded context so the requested scope is applied. + """ + models = list(models) if models is not None else [] + use_project_index = ( + self.config.linter.use_project_index if use_project_index is None else 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 models + } + if models and use_project_index + else None + ) + + # An already-loaded context does not otherwise enter the loading path. Reload when + # indexed linting is requested for specific models so the scope is actually applied. + if not self._loaded or target_fqns is not None: + self.load(model_fqns=target_fqns, use_project_index=use_project_index) + found_error = False model_list = ( diff --git a/sqlmesh/core/loader.py b/sqlmesh/core/loader.py index cb951b4f9e..5a1ce8b17d 100644 --- a/sqlmesh/core/loader.py +++ b/sqlmesh/core/loader.py @@ -3,9 +3,11 @@ import abc import glob import itertools +import json import linecache import os import re +import tempfile import typing as t from collections import Counter, defaultdict from dataclasses import dataclass @@ -38,6 +40,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 +68,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 +192,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 +241,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 +273,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 +302,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 +499,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 +555,170 @@ 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, int, float]]]: + return [ + 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 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") + 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_models in indexed_files.items(): + path = self.config_path / relative_path + 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) + + 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 + + # 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 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: 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, @@ -559,6 +728,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 +740,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 +813,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 +828,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 c625cb084d..a8ee1aaa39 100644 --- a/tests/cli/test_cli.py +++ b/tests/cli/test_cli.py @@ -1352,6 +1352,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 e41382b078..75737f1edb 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 @@ -36,9 +37,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.snapshot import SnapshotChangeCategory @@ -3288,6 +3290,189 @@ 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("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("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_lint_models_project_index_reloads_loaded_context(tmp_path: pathlib.Path) -> 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;") + create_temp_file(tmp_path, pathlib.Path("models", "c.sql"), "MODEL(name c); SELECT 2 AS col;") + + context = Context( + config=Config( + model_defaults=ModelDefaultsConfig(dialect="duckdb"), + linter=LinterConfig(enabled=True, use_project_index=True), + ), + paths=tmp_path, + ) + context.load(use_project_index=True) + loader = t.cast(SqlMeshLoader, context._loaders[0]) + + with patch.object(loader, "_load_sql_models", wraps=loader._load_sql_models) as load_mock: + assert context.lint_models(["b"]) == [] + + assert load_mock.call_count == 1 + selected_paths = load_mock.call_args.kwargs["selected_paths"] + assert {path.name for path in selected_paths} == {"a.sql", "b.sql"} + assert set(context.models) == { + context.get_model(model_name, raise_if_missing=True).fqn for model_name in ("a", "b") + } + + +@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,