Merge branch 'main' into feature/DRM/add-dev-container

Author: davem-bis

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/linter.md

@@ -74,6 +74,7 @@ Here are all of SQLMesh's built-in linting rules:
 | `invalidselectstarexpansion` | Correctness | The query's top-level selection may be `SELECT *`, but only if SQLMesh can expand the `SELECT *` into individual columns |
 | `noselectstar`               | Stylistic   | The query's top-level selection may not be `SELECT *`, even if SQLMesh can expand the `SELECT *` into individual columns |
 | `nomissingaudits`             | Governance  | SQLMesh did not find any `audits` in the model's configuration to test data quality.                                                 |
+| `nomissingunittest` | Governance | SQLMesh did not find any `unit tests` associated with the model to test |
 
 ### User-defined rules
 

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:
 

sqlmesh/core/console.py

@@ -2920,7 +2920,10 @@ def __init__(
 
         super().__init__(console, **kwargs)
 
-        self.display = display or get_ipython().user_ns.get("display", ipython_display)
+        ipython = get_ipython()
+        self.display = display or (
+            ipython.user_ns.get("display", ipython_display) if ipython else ipython_display
+        )
         self.missing_dates_output = widgets.Output()
         self.dynamic_options_after_categorization_output = widgets.VBox()
 

sqlmesh/core/context.py

@@ -118,7 +118,7 @@
     filter_tests_by_patterns,
 )
 from sqlmesh.core.user import User
-from sqlmesh.utils import UniqueKeyDict, Verbosity
+from sqlmesh.utils import CorrelationId, UniqueKeyDict, Verbosity
 from sqlmesh.utils.concurrency import concurrent_apply_to_values
 from sqlmesh.utils.dag import DAG
 from sqlmesh.utils.date import (
@@ -811,6 +811,9 @@ def run(
             engine_type=self.snapshot_evaluator.adapter.dialect,
             state_sync_type=self.state_sync.state_type(),
         )
+        snapshot_evaluator = self.snapshot_evaluator.set_correlation_id(
+            CorrelationId.from_run_id(analytics_run_id)
+        )
         self._load_materializations()
 
         env_check_attempts_num = max(
@@ -863,6 +866,7 @@ def _has_environment_changed() -> bool:
                     select_models=select_models,
                     circuit_breaker=_has_environment_changed,
                     no_auto_upstream=no_auto_upstream,
+                    snapshot_evaluator=snapshot_evaluator,
                 )
                 done = True
             except CircuitBreakerError:
@@ -2605,8 +2609,9 @@ def _run(
         select_models: t.Optional[t.Collection[str]],
         circuit_breaker: t.Optional[t.Callable[[], bool]],
         no_auto_upstream: bool,
+        snapshot_evaluator: t.Optional[SnapshotEvaluator] = None,
     ) -> CompletionStatus:
-        scheduler = self.scheduler(environment=environment)
+        scheduler = self.scheduler(environment=environment, snapshot_evaluator=snapshot_evaluator)
         snapshots = scheduler.snapshots
 
         if select_models is not None:
@@ -3065,10 +3070,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,
                     )
@@ -3080,6 +3092,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:

sqlmesh/core/dialect.py

@@ -551,6 +551,36 @@ def _parse_table_parts(
     return table
 
 
+# Only needed for T-SQL: it spells a column's nullability right after its type, e.g.
+# ALTER TABLE t ALTER COLUMN c INT NOT NULL. Without this the trailing clause is left
+# over, so the whole statement falls back to a Command and any macros it contains (such as
+# @this_model) are no longer resolved, which means they reach the engine verbatim.
+#
+# See: https://learn.microsoft.com/en-us/sql/t-sql/statements/alter-table-transact-sql
+def _parse_alter_table_alter(self: Parser) -> t.Optional[exp.Expr]:
+    alter_column = self.__parse_alter_table_alter()  # type: ignore
+
+    if isinstance(alter_column, exp.AlterColumn) and alter_column.args.get("dtype"):
+        if self._match_pair(TokenType.NOT, TokenType.NULL):
+            alter_column.set("allow_null", False)
+        elif self._match(TokenType.NULL):
+            alter_column.set("allow_null", True)
+
+    return alter_column
+
+
+def altercolumn_sql(self: Generator, expression: exp.AlterColumn) -> str:
+    sql = self._altercolumn_sql(expression)  # type: ignore
+
+    # sqlglot's generator returns as soon as it renders the type, so the nullability parsed
+    # above has to be appended here
+    allow_null = expression.args.get("allow_null")
+    if expression.args.get("dtype") and allow_null is not None:
+        sql = f"{sql} NULL" if allow_null else f"{sql} NOT NULL"
+
+    return sql
+
+
 def _parse_if(self: Parser) -> t.Optional[exp.Expr]:
     # If we fail to parse an IF function with expressions as arguments, we then try
     # to parse a statement / command to support the macro @IF(condition, statement)
@@ -780,8 +810,12 @@ def _parse_interval_span(self: Parser, this: exp.Expr) -> exp.Interval:
     return interval
 
 
-def _override(klass: t.Type[Tokenizer | Parser], func: t.Callable) -> None:
+def _override(klass: t.Type[Tokenizer | Parser | Generator], func: t.Callable) -> None:
     name = func.__name__
+    if getattr(klass, name, None) is func:
+        # Already overridden. Re-applying would save the override itself as the
+        # "original", making the wrapper call itself and recurse infinitely.
+        return
     setattr(klass, f"_{name}", getattr(klass, name))
     setattr(klass, name, func)
 
@@ -1170,11 +1204,12 @@ def extend_sqlglot() -> None:
                 MacroDef,
             )
 
-        generator.UNWRAPPED_INTERVAL_VALUES = (
-            *generator.UNWRAPPED_INTERVAL_VALUES,
-            MacroStrReplace,
-            MacroVar,
-        )
+        if MacroVar not in generator.UNWRAPPED_INTERVAL_VALUES:
+            generator.UNWRAPPED_INTERVAL_VALUES = (
+                *generator.UNWRAPPED_INTERVAL_VALUES,
+                MacroStrReplace,
+                MacroVar,
+            )
 
     _override(Parser, _parse_select)
     _override(Parser, _parse_statement)
@@ -1194,6 +1229,8 @@ def extend_sqlglot() -> None:
     _override(Parser, _parse_interval_span)
     _override(Parser, _warn_unsupported)
     _override(Snowflake.Parser, _parse_table_parts)
+    _override(TSQL.Parser, _parse_alter_table_alter)
+    _override(TSQL.Generator, altercolumn_sql)
 
     # DuckDB's prefix absolute power operator `@` clashes with the macro syntax
     DuckDB.Parser.NO_PAREN_FUNCTION_PARSERS.pop("@", None)

sqlmesh/core/engine_adapter/base.py

@@ -178,6 +178,7 @@ def with_settings(self, **kwargs: t.Any) -> EngineAdapter:
             "query_execution_tracker": kwargs.pop(
                 "query_execution_tracker", self._query_execution_tracker
             ),
+            "pre_ping": kwargs.pop("pre_ping", self._pre_ping),
             **self._extra_config,
             **kwargs,
         }

sqlmesh/core/engine_adapter/clickhouse.py

@@ -595,6 +595,48 @@ def _create_table(
                 target_columns_to_types or self.columns(table_name),
             )
 
+    def create_view(
+        self,
+        view_name: TableName,
+        query_or_df: QueryOrDF,
+        target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
+        replace: bool = True,
+        materialized: bool = False,
+        materialized_properties: t.Optional[t.Dict[str, t.Any]] = None,
+        table_description: t.Optional[str] = None,
+        column_descriptions: t.Optional[t.Dict[str, str]] = None,
+        view_properties: t.Optional[t.Dict[str, exp.Expr]] = None,
+        source_columns: t.Optional[t.List[str]] = None,
+        **create_kwargs: t.Any,
+    ) -> None:
+        if self._default_catalog and isinstance(query_or_df, exp.Query):
+            from sqlmesh.utils.errors import SQLMeshError
+
+            query_or_df = query_or_df.copy()
+            for table in query_or_df.find_all(exp.Table):
+                if not table.catalog:
+                    continue
+                if table.catalog != self._default_catalog:
+                    raise SQLMeshError(
+                        f"{self.dialect} requires that all catalog operations be against a single "
+                        f"catalog: {self._default_catalog}. Provided catalog: {table.catalog}"
+                    )
+                table.set("catalog", None)
+
+        super().create_view(
+            view_name,
+            query_or_df,
+            target_columns_to_types=target_columns_to_types,
+            replace=replace,
+            materialized=materialized,
+            materialized_properties=materialized_properties,
+            table_description=table_description,
+            column_descriptions=column_descriptions,
+            view_properties=view_properties,
+            source_columns=source_columns,
+            **create_kwargs,
+        )
+
     def _strip_virtual_catalog(self, name: "TableName") -> exp.Table:
         """Strip the virtual catalog prefix from a table name if present.
 
@@ -850,7 +892,7 @@ def _build_table_properties_exp(
             primary_key_vals = []
             if isinstance(primary_key, (exp.Tuple, exp.Array)):
                 primary_key_vals = primary_key.expressions
-            if isinstance(ordered_by_raw, exp.Paren):
+            if isinstance(primary_key, exp.Paren):
                 primary_key_vals = [primary_key.this]
 
             if not primary_key_vals: