Merge branch 'main' into lint-scope-schema-resolution

Author: fresioAS

.github/workflows/pr.yaml

@@ -514,7 +514,12 @@ jobs:
             echo "DBT version is ${{ matrix.dbt-version }} (>= 1.6.0), keeping semantic_models and metrics sections"
           fi
 
-          make dbt-fast-test
+          # dbt 1.6 installs without web/lsp (pydantic conflict); skip collecting those tests
+          if [[ "${{ matrix.dbt-version }}" == "1.6" ]]; then
+            pytest -n auto -m "dbt and fast" --reruns 3 --ignore=tests/lsp --ignore=tests/web
+          else
+            make dbt-fast-test
+          fi
       - name: Test SQLMesh info in sushi_dbt
         working-directory: ./examples/sushi_dbt
         run: |

CONTRIBUTING.md

@@ -14,7 +14,7 @@ Initial TSC voting members are the project's Maintainers:
 |---------------------|---------------|----------------|------------| 
 | Alexander Butler    | z3z1ma        | Harness        | TSC Member |
 | Alexander Filipchik | afilipchik    | Cloud Kitchens | TSC Member |
-| Reid Hooper         | rhooper9711   | Benzinga       | TSC Member |
+| Cortland Goffena    | cmgoffena13   | Benzinga       | TSC Member |
 | Yuki Kakegawa       | StuffbyYuki   | Jump.ai        | TSC Member |
 | Toby Mao            | tobymao       | Fivetran       | TSC Chair  |
 | Alex Wilde          | alexminerv    | Minerva        | TSC Member |

Makefile

@@ -46,7 +46,12 @@ install-dev-dbt-%:
 		echo "Applying numpy<2 constraint for dbt $$version"; \
 		$(SED_INPLACE) 's/"numpy"/"numpy<2"/g' pyproject.toml; \
 	fi; \
-	$(MAKE) install-dev; \
+	if [ "$$version" = "1.6.0" ]; then \
+		echo "Installing without web/lsp for dbt 1.6.0 (fastapi>=0.136 needs pydantic v2; dbt 1.6 needs pydantic v1)"; \
+		$(PIP) install -e ".[dev,slack,dlt]" ./examples/custom_materializations; \
+	else \
+		$(MAKE) install-dev; \
+	fi; \
 	if [ "$$version" = "1.6.0" ]; then \
 		echo "Applying overrides for dbt 1.6.0"; \
 		$(PIP) install 'pydantic>=2.0.0' 'google-cloud-bigquery==3.30.0' 'databricks-sdk==0.28.0' \

docs/concepts/macros/sqlmesh_macros.md

@@ -1608,7 +1608,7 @@ def add_args(
     return argument_1 + argument_2 + argument_3
 ```
 
-An `@add_args` call providing values for all arguments accepts positional arguments like this: `@add_args(5, 6, 7)` (which returns 5 + 6 + 7 = `18`). A call omitting and using the default value for the the final `argument_3` can also use positional arguments: `@add_args(5, 6)` (which returns 5 + 6 + 3 = `14`).
+An `@add_args` call providing values for all arguments accepts positional arguments like this: `@add_args(5, 6, 7)` (which returns 5 + 6 + 7 = `18`). A call omitting and using the default value for the final `argument_3` can also use positional arguments: `@add_args(5, 6)` (which returns 5 + 6 + 3 = `14`).
 
 However, skipping an argument requires specifying the names of subsequent arguments (i.e., using "keyword arguments"). For example, skipping the second argument above by just omitting it - `@add_args(5, , 7)` - results in an error.
 

docs/concepts/models/model_kinds.md

@@ -583,7 +583,7 @@ MODEL (
 
 ### When Matched Expression
 
-The logic to use when updating columns when a match occurs (the source and target match on the given keys) by default updates all the columns. This can be overriden with custom logic like below:
+The logic to use when updating columns when a match occurs (the source and target match on the given keys) by default updates all the columns. This can be overridden with custom logic like below:
 
 ```sql linenums="1" hl_lines="5"
 MODEL (
@@ -1437,7 +1437,7 @@ GROUP BY
 
 SCD Type 2 models are designed by default to protect the data that has been captured because it is not possible to recreate the history once it has been lost.
 However, there are cases where you may want to clear the history and start fresh.
-For this use use case you will want to start by setting `disable_restatement` to `false` in the model definition.
+For this use case you will want to start by setting `disable_restatement` to `false` in the model definition.
 
 ```sql linenums="1" hl_lines="5"
 MODEL (

docs/guides/ui.md

@@ -225,7 +225,7 @@ You may include all a project's models by clicking `All` in the Show drop-down o
 
 ![Lineage module - all models](./ui/ui-guide_lineage-all.png){ loading=lazy }
 
-Click `Connected` in the Show drop-down menu to highlight edges between upstream parents and downstream children in blue. This may be helpful when when a project contains many models:
+Click `Connected` in the Show drop-down menu to highlight edges between upstream parents and downstream children in blue. This may be helpful when a project contains many models:
 
 ![Lineage module - all models, connected edges](./ui/ui-guide_lineage-all-connected.png){ loading=lazy }
 

docs/quickstart/notebook.md

@@ -248,7 +248,7 @@ You've now created a new production environment with all of history backfilled.
 
 ## 3. Update a model
 
-Now that we have have populated the `prod` environment, let's modify one of the SQL models.
+Now that we have populated the `prod` environment, let's modify one of the SQL models.
 
 We can modify the incremental SQL model using the `%model` *line* notebook magic (note the single `%`) and the model name:
 

pyproject.toml

@@ -134,15 +134,19 @@ snowflake = [
 starrocks = ["pymysql"]
 trino = ["trino"]
 web = [
-    "fastapi==0.120.1",
+    "fastapi>=0.136.3; python_version >= '3.10'",
+    "starlette>=1.0.1; python_version >= '3.10'",
+    "fastapi==0.120.1; python_version < '3.10'",
     "watchfiles>=0.19.0",
     "uvicorn[standard]==0.22.0",
     "sse-starlette>=0.2.2",
     "pyarrow",
 ]
 lsp = [
     # Duplicate of web
-    "fastapi==0.120.1",
+    "fastapi>=0.136.3; python_version >= '3.10'",
+    "starlette>=1.0.1; python_version >= '3.10'",
+    "fastapi==0.120.1; python_version < '3.10'",
     "watchfiles>=0.19.0",
     # "uvicorn[standard]==0.22.0",
     "sse-starlette>=0.2.2",

sqlmesh/cli/main.py

@@ -645,7 +645,7 @@ def run(ctx: click.Context, environment: t.Optional[str] = None, **kwargs: t.Any
 def invalidate(ctx: click.Context, environment: str, **kwargs: t.Any) -> None:
     """Invalidate the target environment, forcing its removal during the next run of the janitor process."""
     context = ctx.obj
-    context.invalidate_environment(environment, **kwargs)
+    context.invalidate_environment(environment, must_exist=True, **kwargs)
 
 
 @cli.command("janitor")

sqlmesh/core/context.py

@@ -1938,15 +1938,24 @@ def apply(
         )
 
     @python_api_analytics
-    def invalidate_environment(self, name: str, sync: bool = False) -> None:
+    def invalidate_environment(
+        self, name: str, sync: bool = False, must_exist: bool = False
+    ) -> None:
         """Invalidates the target environment by setting its expiration timestamp to now.
 
         Args:
             name: The name of the environment to invalidate.
             sync: If True, the call blocks until the environment is deleted. Otherwise, the environment will
                 be deleted asynchronously by the janitor process.
+            must_exist: If True, raise if the environment doesn't exist instead of silently doing nothing.
+                Used by the user-facing entry points, where a mistyped name should be reported rather than
+                look like it succeeded. Internal callers such as
+                `GithubController.try_invalidate_pr_environment` rely on the default no-op behavior, since
+                a PR environment may never have been created.
         """
         name = Environment.sanitize_name(name)
+        if must_exist and self.state_sync.get_environment(name) is None:
+            raise SQLMeshError(f"Environment '{name}' was not found.")
         self.state_sync.invalidate_environment(name)
         if sync:
             self._cleanup_environments(name=name)
@@ -3138,10 +3147,17 @@ def _cleanup_environments(
             expired_env = self.state_reader.get_environment(expired_env_summary.name)
 
             if expired_env:
+                cleanup_default_adapter, cleanup_engine_adapters, failure = (
+                    self._cleanup_adapters_for_environment(expired_env)
+                )
+                if failure:
+                    logger.warning(failure)
+                    failures.append(failure)
+                    continue
                 failures.extend(
                     cleanup_expired_views(
-                        default_adapter=self.engine_adapter,
-                        engine_adapters=self.engine_adapters,
+                        default_adapter=cleanup_default_adapter,
+                        engine_adapters=cleanup_engine_adapters,
                         environments=[expired_env],
                         console=self.console,
                     )
@@ -3153,6 +3169,62 @@ def _cleanup_environments(
             self.state_sync.delete_expired_environments(current_ts=current_ts, name=name)
         return failures
 
+    def _cleanup_adapters_for_environment(
+        self, environment: Environment
+    ) -> t.Tuple[EngineAdapter, t.Dict[str, EngineAdapter], t.Optional[str]]:
+        """Create cleanup-scoped adapters for an expired environment.
+
+        Persisted catalog-qualified view names indicate that virtual catalog injection was active,
+        so cleanup can clone only the selected adapters with the historical catalog and leave the
+        context's adapters unchanged.
+        """
+        engine_adapters = self.engine_adapters
+        default_adapter = self.engine_adapter
+        catalogs_by_gateway: t.Dict[str, t.Set[str]] = collections.defaultdict(set)
+
+        for snapshot in environment.snapshots:
+            if not snapshot.is_model or snapshot.is_symbolic:
+                continue
+
+            gateway = (
+                snapshot.model_gateway
+                if environment.gateway_managed and snapshot.model_gateway in engine_adapters
+                else self.selected_gateway
+            )
+            adapter = engine_adapters.get(gateway, default_adapter)
+            catalog = snapshot.qualified_view_name.catalog_for_environment(
+                environment.naming_info, dialect=adapter.dialect
+            )
+            if catalog and adapter.supports_virtual_catalog() is True:
+                catalogs_by_gateway[gateway].add(catalog)
+
+        for gateway, catalogs in catalogs_by_gateway.items():
+            if len(catalogs) > 1:
+                catalogs_description = ", ".join(f"'{catalog}'" for catalog in sorted(catalogs))
+                return (
+                    default_adapter,
+                    engine_adapters,
+                    (
+                        f"Failed to clean up expired environment '{environment.name}': gateway "
+                        f"'{gateway}' references multiple virtual catalogs: {catalogs_description}"
+                    ),
+                )
+
+        cleanup_engine_adapters = engine_adapters.copy()
+        cleanup_default_adapter = default_adapter
+        for gateway, catalogs in catalogs_by_gateway.items():
+            cleanup_adapter = engine_adapters.get(gateway, default_adapter).with_settings()
+            cleanup_adapter.inject_virtual_catalog(gateway)
+            # inject_virtual_catalog() may initialize adapter-specific state in addition to
+            # _default_catalog. Override only the cleanup clone with the catalog persisted in the
+            # expired environment so historical names pass SINGLE_CATALOG_ONLY validation.
+            cleanup_adapter._default_catalog = next(iter(catalogs))
+            cleanup_engine_adapters[gateway] = cleanup_adapter
+            if gateway == self.selected_gateway:
+                cleanup_default_adapter = cleanup_adapter
+
+        return cleanup_default_adapter, cleanup_engine_adapters, None
+
     def _try_connection(self, connection_name: str, validator: t.Callable[[], None]) -> None:
         connection_name = connection_name.capitalize()
         try: