Merge branch 'main' into lint-model-paths

Author: tripleaceme

.github/workflows/pr.yaml

@@ -558,7 +558,8 @@ jobs:
         run: |
           source ../../.venv/bin/activate
           sed -i 's/target: in_memory/target: postgres/g' profiles.yml
-          if [[ $(echo -e "${{ matrix.dbt-version }}\n1.5.0" | sort -V | head -n1) == "${{ matrix.dbt-version }}" ]] && [[ "${{ matrix.dbt-version }}" != "1.5.0" ]]; then
+          if [[ "${{ matrix.dbt-version }}" == "1.3" ]] || \
+             [[ "${{ matrix.dbt-version }}" == "1.4" ]]; then
             echo "DBT version is ${{ matrix.dbt-version }} (< 1.5.0), removing version parameters..."
             sed -i -e 's/, version=1) }}/) }}/g' -e 's/, v=1) }}/) }}/g' models/top_waiters.sql
           else

docs/concepts/tests.md

@@ -463,6 +463,20 @@ You can also run tests that match a pattern or substring using a glob pathname e
 $ sqlmesh test tests/test_*
 ```
 
+Passing the path of a model file runs the tests for that model, which is useful for commit hooks and other tools that work with changed files rather than test names:
+
+```
+$ sqlmesh test models/full_model.sql
+```
+
+Model files and test files can be mixed, and the results are unioned. A test selected by more than one argument still runs only once, so the following runs each of `full_model`'s tests a single time even though both arguments cover them:
+
+```
+$ sqlmesh test models/full_model.sql tests/test_full_model.yaml
+```
+
+An argument that is neither a known model file nor a known test file is an error, so a mistyped or stale path fails instead of quietly running no tests. A model that simply has no tests is not an error.
+
 You can pass `--local` to run tests without loading state from the configured state connection:
 
 ``` bash

docs/reference/cli.md

@@ -623,6 +623,10 @@ Usage: sqlmesh test [OPTIONS] [TESTS]...
 
   Run model unit tests.
 
+  TESTS are test files, `file.yaml::test_name` selectors, or model files, in
+  which case the tests for those models are run. They are unioned, and a test
+  selected more than once still only runs once.
+
 Options:
   -k TEXT              Only run tests that match the pattern of substring.
   -v, --verbose        Verbose output.

pnpm-lock.yaml

@@ -835,7 +835,12 @@ def test(
     select_model: t.List[str],
     tests: t.List[str],
 ) -> None:
-    """Run model unit tests."""
+    """Run model unit tests.
+
+    TESTS are test files, `file.yaml::test_name` selectors, or model files, in which case the
+    tests for those models are run. They are unioned, and a test selected more than once still
+    only runs once.
+    """
     model_names = (
         obj._new_selector().expand_model_selections(select_model) if select_model else None
     )
@@ -845,6 +850,7 @@ def test(
         verbosity=Verbosity(verbose),
         preserve_fixtures=preserve_fixtures,
         model_names=model_names,
+        raise_on_unknown_paths=True,
     )
     if not result.wasSuccessful():
         exit(1)

sqlmesh/cli/main.py

@@ -36,6 +36,7 @@
 import abc
 import collections
 import logging
+import os.path
 import sys
 import time
 import traceback
@@ -107,6 +108,7 @@
     CachingStateSync,
     StateReader,
     StateSync,
+    Versions,
 )
 from sqlmesh.core.janitor import cleanup_expired_views, delete_expired_snapshots
 from sqlmesh.core.table_diff import TableDiff
@@ -119,7 +121,7 @@
     filter_tests_by_patterns,
 )
 from sqlmesh.core.user import User
-from sqlmesh.utils import CorrelationId, UniqueKeyDict, Verbosity
+from sqlmesh.utils import CorrelationId, UniqueKeyDict, Verbosity, unique
 from sqlmesh.utils.concurrency import concurrent_apply_to_values
 from sqlmesh.utils.dag import DAG
 from sqlmesh.utils.date import (
@@ -2435,17 +2437,26 @@ def test(
         preserve_fixtures: bool = False,
         stream: t.Optional[t.TextIO] = None,
         model_names: t.Optional[t.Collection[str]] = None,
+        raise_on_unknown_paths: bool = False,
     ) -> ModelTextTestResult:
         """Discover and run model tests"""
         if verbosity >= Verbosity.VERBOSE:
             import pandas as pd
 
             pd.set_option("display.max_columns", None)
 
-        baseline_meta = self.select_tests(tests=tests, patterns=match_patterns, model_names=None)
+        baseline_meta = self.select_tests(
+            tests=tests,
+            patterns=match_patterns,
+            model_names=None,
+            raise_on_unknown_paths=raise_on_unknown_paths,
+        )
         if model_names is not None:
             test_meta = self.select_tests(
-                tests=tests, patterns=match_patterns, model_names=model_names
+                tests=tests,
+                patterns=match_patterns,
+                model_names=model_names,
+                raise_on_unknown_paths=raise_on_unknown_paths,
             )
             tests_skipped = len(baseline_meta) - len(test_meta)
         else:
@@ -2628,15 +2639,18 @@ def migrate(self) -> None:
         """
         self.notification_target_manager.notify(NotificationEvent.MIGRATION_START)
         self._load_materializations()
+        state_sync = self._new_state_sync()
+        previous_versions = self._state_versions(state_sync)
         try:
-            self._new_state_sync().migrate(
+            state_sync.migrate(
                 promoted_snapshots_only=self.config.migration.promoted_snapshots_only,
             )
         except Exception as e:
             self.notification_target_manager.notify(
                 NotificationEvent.MIGRATION_FAILURE, traceback.format_exc()
             )
             raise e
+        self._print_state_versions(self._state_versions(state_sync), previous_versions)
         self.notification_target_manager.notify(NotificationEvent.MIGRATION_END)
 
     @python_api_analytics
@@ -2645,7 +2659,10 @@ def rollback(self) -> None:
 
         Please contact your SQLMesh administrator before doing this. This action cannot be undone.
         """
-        self._new_state_sync().rollback()
+        state_sync = self._new_state_sync()
+        previous_versions = self._state_versions(state_sync)
+        state_sync.rollback()
+        self._print_state_versions(self._state_versions(state_sync), previous_versions)
 
     @python_api_analytics
     def create_external_models(self, strict: bool = False) -> None:
@@ -2719,6 +2736,12 @@ def print_info(
         if state_connection:
             self._try_connection("state backend", state_connection.connection_validator())
 
+        if verbosity >= Verbosity.VERBOSE:
+            try:
+                self._print_state_versions(self._state_versions())
+            except Exception as ex:
+                self.console.log_error(f"Failed to fetch the state backend versions. {ex}")
+
     @python_api_analytics
     def print_environment_names(self) -> None:
         """Prints all environment names along with expiry datetime."""
@@ -3308,6 +3331,25 @@ def _try_connection(self, connection_name: str, validator: t.Callable[[], None])
         except Exception as ex:
             self.console.log_error(f"{connection_name} connection failed. {ex}")
 
+    def _state_versions(self, state_sync: t.Optional[StateSync] = None) -> Versions:
+        """Returns the versions recorded in the state backend without validating them."""
+        return (state_sync or self._new_state_sync()).get_versions(validate=False)
+
+    def _print_state_versions(
+        self, versions: Versions, previous_versions: t.Optional[Versions] = None
+    ) -> None:
+        """Prints the state backend versions, optionally alongside the ones they replaced."""
+        self.console.log_status_update("\nState backend versions:")
+        for label, attribute in (
+            ("Schema version", "schema_version"),
+            ("SQLGlot version", "sqlglot_version"),
+            ("SQLMesh version", "sqlmesh_version"),
+        ):
+            version = getattr(versions, attribute)
+            if previous_versions is not None:
+                version = f"{getattr(previous_versions, attribute)} -> {version}"
+            self.console.log_status_update(f"{label}: {version}")
+
     def _new_state_sync(self) -> StateSync:
         return self._provided_state_sync or self._scheduler.create_state_sync(self)
 
@@ -3704,30 +3746,112 @@ def lint_models(
 
         return all_violations
 
+    def _tests_by_absolute_model_path(self) -> t.Dict[str, t.List[ModelTestMetadata]]:
+        """Map each model file to the tests that target the model(s) defined in it."""
+        tests_by_model_name: t.Dict[str, t.List[ModelTestMetadata]] = collections.defaultdict(list)
+        for metadata in self._model_test_metadata:
+            if metadata.model_name:
+                tests_by_model_name[
+                    normalize_model_name(
+                        metadata.model_name,
+                        default_catalog=self.default_catalog,
+                        dialect=self.default_dialect,
+                    )
+                ].append(metadata)
+
+        # A path is made absolute rather than resolved, so this costs no syscalls per model.
+        tests_by_path: t.Dict[str, t.List[ModelTestMetadata]] = {}
+        for fqn, model in self._models.items():
+            if model._path is not None:
+                tests_by_path.setdefault(os.path.abspath(model._path), []).extend(
+                    tests_by_model_name.get(fqn, [])
+                )
+
+        return tests_by_path
+
+    def _select_tests_by_test_path(self, selector: str) -> t.Optional[t.List[ModelTestMetadata]]:
+        """Resolve a selector against the test files, or return None if it matches none of them.
+
+        The selector is a test file path or a `path::test_name`. Paths are matched as given
+        first, so an unchanged selector never pays for normalization.
+        """
+        if "::" in selector:
+            metadata = self._model_test_metadata_fully_qualified_name_index.get(selector)
+            if metadata is None:
+                path, _, test_name = selector.rpartition("::")
+                metadata = self._model_test_metadata_fully_qualified_name_index.get(
+                    f"{os.path.abspath(path)}::{test_name}"
+                )
+            return [metadata] if metadata is not None else None
+
+        for candidate in (Path(selector), Path(os.path.abspath(selector))):
+            matched = self._model_test_metadata_path_index.get(candidate)
+            if matched is not None:
+                return list(matched)
+
+        return None
+
+    def _unknown_test_selector_error(self, selector: str) -> str:
+        """Explains why a selector matched nothing.
+
+        A `path::test_name` whose file is a known test file failed on the test name, not the
+        path, so the message says so rather than claiming the file is unknown.
+        """
+        if "::" in selector:
+            path, _, _ = selector.rpartition("::")
+            if any(
+                candidate in self._model_test_metadata_path_index
+                for candidate in (Path(path), Path(os.path.abspath(path)))
+            ):
+                return f"'{selector}' is not a known test in '{path}'."
+
+        return f"'{selector}' is not a known model or test file."
+
     def select_tests(
         self,
         tests: t.Optional[t.List[str]] = None,
         patterns: t.Optional[t.List[str]] = None,
         model_names: t.Optional[t.Collection[str]] = None,
+        raise_on_unknown_paths: bool = False,
     ) -> t.List[ModelTestMetadata]:
-        """Filter pre-loaded test metadata based on tests and patterns."""
+        """Filter pre-loaded test metadata based on tests and patterns.
+
+        Args:
+            tests: Test selectors. Each one is a test file path, a `path::test_name`, or the path
+                of a model file, in which case that model's tests are selected. Selectors are
+                unioned and the result is deduplicated, so a model file and a test file that
+                resolve to the same test run it once rather than twice.
+            patterns: Patterns matched against fully qualified test names.
+            model_names: If given, narrows the selection to tests targeting these models.
+            raise_on_unknown_paths: Whether to raise when a selector matches neither a known test
+                nor a known model file. Off by default so that callers which probe arbitrary
+                documents, such as the LSP, keep getting an empty result instead of an error.
+        """
 
         test_meta = self._model_test_metadata
 
         if tests:
-            filtered_tests = []
+            filtered_tests: t.List[ModelTestMetadata] = []
+            # Built at most once, and only if a selector turns out not to be a test file.
+            tests_by_model_path: t.Optional[t.Dict[str, t.List[ModelTestMetadata]]] = None
+
             for test in tests:
-                if "::" in test:
-                    if test in self._model_test_metadata_fully_qualified_name_index:
-                        filtered_tests.append(
-                            self._model_test_metadata_fully_qualified_name_index[test]
-                        )
-                else:
-                    test_path = Path(test)
-                    if test_path in self._model_test_metadata_path_index:
-                        filtered_tests.extend(self._model_test_metadata_path_index[test_path])
+                matched = self._select_tests_by_test_path(test)
+                if matched is None and "::" not in test:
+                    if tests_by_model_path is None:
+                        tests_by_model_path = self._tests_by_absolute_model_path()
+                    # A known model with no tests matches an empty list, which is not the same
+                    # as a selector that resolves to nothing at all.
+                    matched = tests_by_model_path.get(os.path.abspath(test))
+                if matched is None:
+                    if raise_on_unknown_paths:
+                        raise SQLMeshError(self._unknown_test_selector_error(test))
+                    continue
+                filtered_tests.extend(matched)
 
-            test_meta = filtered_tests
+            # Selectors can overlap, e.g. a model file and the test file holding its tests, so
+            # the union is deduplicated to avoid running the same test more than once.
+            test_meta = unique(filtered_tests)
 
         if patterns:
             test_meta = filter_tests_by_patterns(test_meta, patterns)

sqlmesh/core/context.py

@@ -180,7 +180,10 @@ def drop_schema(
         **drop_args: t.Dict[str, exp.Expr],
     ) -> None:
         """
-        MsSql doesn't support CASCADE clause and drops schemas unconditionally.
+        MsSql doesn't support CASCADE clause so objects are dropped individually.
+
+        SQL Server also forbids dropping the built-in ``dbo`` schema (error 15150).
+        Objects inside it are still dropped when cascade=True, but the schema itself is left in place.
         """
         if cascade:
             objects = self._get_data_objects(schema_name)
@@ -199,6 +202,15 @@ def drop_schema(
                         object_table,
                         exists=ignore_if_not_exists,
                     )
+
+        schema = (
+            (schema_name.db or schema_name.name)
+            if isinstance(schema_name, exp.Table)
+            else schema_name
+        )
+        if schema.lower() == "dbo":
+            return
+
         super().drop_schema(schema_name, ignore_if_not_exists=ignore_if_not_exists, cascade=False)
 
     def merge(

sqlmesh/core/engine_adapter/mssql.py

@@ -96,6 +96,15 @@ def migrate(
             migrate_rows = self._apply_migrations(schema, skip_backup)
 
             if not migrate_rows and major_minor(SQLMESH_VERSION) == versions.minor_sqlmesh_version:
+                # Nothing to migrate, but a patch-level bump still leaves the recorded versions
+                # behind what is actually running, so they are brought up to date here. The
+                # schema version is carried over rather than defaulted, since no migration ran
+                # and moving it could hide one that is genuinely needed later.
+                if (
+                    versions.sqlmesh_version != SQLMESH_VERSION
+                    or versions.sqlglot_version != SQLGLOT_VERSION
+                ):
+                    self.version_state.update_versions(schema_version=versions.schema_version)
                 return
 
             if migrate_rows: