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..2ea74259ab 100644 --- a/docs/reference/cli.md +++ b/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. -``` \ No newline at end of file +``` diff --git a/sqlmesh/cli/main.py b/sqlmesh/cli/main.py index f5a3c62cf9..1aeb05ee62 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 + # 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) diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py index f30c401a36..512e831bc2 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, @@ -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 = ( 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 092def8e0c..9a7f702bfa 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) @@ -2028,6 +2073,19 @@ def test_render(runner: CliRunner, tmp_path: Path): assert expected in cleaned_output + indexed_result = runner.invoke( + cli, + [ + "--paths", + str(tmp_path), + "render", + "sqlmesh_example.full_model", + "--use-project-index", + "--no-format", + ], + ) + assert indexed_result.exit_code == 0 + @time_machine.travel(FREEZE_TIME) def test_signals(runner: CliRunner, tmp_path: Path): diff --git a/tests/core/test_context.py b/tests/core/test_context.py index 74ce0d7916..162eb73cd1 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.renderer import render_statements @@ -273,6 +275,82 @@ def test_render_seed_model(sushi_context, assert_exp_eq): ) +@pytest.mark.slow +def test_render_only_creates_snapshots_for_upstream_models(sushi_context: Context): + model = sushi_context.get_model("sushi.top_waiters", raise_if_missing=True) + upstream_fqns = {model.fqn, *sushi_context.dag.upstream(model.fqn)} + + # Sanity check that the project contains models outside of the target model's subgraph. + assert set(sushi_context.models) - upstream_fqns + + with patch.object( + sushi_context.state_reader, + "get_snapshots", + wraps=sushi_context.state_reader.get_snapshots, + ) as default_get_snapshots_mock: + sushi_context.render("sushi.top_waiters") + + default_requested_names = { + snapshot.name + for call_args in default_get_snapshots_mock.call_args_list + for snapshot in call_args.args[0] + } + assert set(sushi_context.models) <= default_requested_names + + with patch.object( + sushi_context.state_reader, + "get_snapshots", + wraps=sushi_context.state_reader.get_snapshots, + ) as get_snapshots_mock: + sushi_context.render("sushi.top_waiters", use_project_index=True) + + requested_names = { + snapshot.name + for call_args in get_snapshots_mock.call_args_list + for snapshot in call_args.args[0] + } + assert model.fqn in requested_names + assert requested_names == upstream_fqns + + +def test_render_only_loads_upstream_model_files(tmp_path: pathlib.Path) -> None: + create_temp_file( + tmp_path, + pathlib.Path("models", "a.sql"), + "MODEL(name a, kind FULL); SELECT 1 AS col;", + ) + create_temp_file( + tmp_path, + pathlib.Path("models", "b.sql"), + "MODEL(name b, kind FULL); SELECT col FROM a;", + ) + create_temp_file( + tmp_path, + pathlib.Path("models", "c.sql"), + "MODEL(name c, kind FULL); SELECT col FROM b;", + ) + config = Config(model_defaults=ModelDefaultsConfig(dialect="duckdb")) + + # Populate the persistent model path/dependency index. + Context(config=config, paths=tmp_path, load=False).load(use_project_index=True) + + ctx = Context(config=config, paths=tmp_path, load=False) + loader = t.cast(SqlMeshLoader, ctx._loaders[0]) + with patch.object( + loader, + "_load_sql_models", + wraps=loader._load_sql_models, + ) as load_sql_models_mock: + ctx.render("b", use_project_index=True) + + 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, + } + + @pytest.mark.slow def test_diff(sushi_context: Context, mocker: MockerFixture): mock_console = mocker.Mock() @@ -2937,6 +3015,163 @@ 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) + + +@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,