Merge branch 'main' into fix/databricks-oauth-shared-connection

Author: gabepesco

sqlmesh/core/config/connection.py

@@ -34,6 +34,7 @@
     ValidationInfo,
     field_validator,
     model_validator,
+    validation_data,
     validation_error_message,
     get_concrete_types_from_typehint,
 )
@@ -1083,7 +1084,7 @@ def validate_execution_project(
         v: t.Optional[str],
         info: ValidationInfo,
     ) -> t.Optional[str]:
-        if v and not info.data.get("project"):
+        if v and not validation_data(info).get("project"):
             raise ConfigError(
                 "If the `execution_project` field is specified, you must also specify the `project` field to provide a default object location."
             )
@@ -1095,7 +1096,7 @@ def validate_quota_project(
         v: t.Optional[str],
         info: ValidationInfo,
     ) -> t.Optional[str]:
-        if v and not info.data.get("project"):
+        if v and not validation_data(info).get("project"):
             raise ConfigError(
                 "If the `quota_project` field is specified, you must also specify the `project` field to provide a default object location."
             )

sqlmesh/core/context.py

@@ -1605,9 +1605,11 @@ def plan_builder(
             backfill_models = None
 
         models_override: t.Optional[UniqueKeyDict[str, Model]] = None
+        selected_fqns: t.Set[str] = set()
+        selected_deletion_fqns: t.Set[str] = set()
         if select_models:
             try:
-                models_override = model_selector.select_models(
+                models_override, selected_fqns = model_selector.select_models(
                     select_models,
                     environment,
                     fallback_env_name=create_from or c.PROD,
@@ -1622,12 +1624,17 @@ def plan_builder(
                 # Only backfill selected models unless explicitly specified.
                 backfill_models = model_selector.expand_model_selections(select_models)
 
+            if not backfill_models:
+                # The selection matched nothing locally. Check whether it matched models
+                # in the deployed environment that were deleted locally.
+                selected_deletion_fqns = selected_fqns - set(self._models)
+
         expanded_restate_models = None
         if restate_models is not None:
             expanded_restate_models = model_selector.expand_model_selections(restate_models)
 
         if (restate_models is not None and not expanded_restate_models) or (
-            backfill_models is not None and not backfill_models
+            backfill_models is not None and not backfill_models and not selected_deletion_fqns
         ):
             raise PlanError(
                 "Selector did not return any models. Please check your model selection and try again."
@@ -1636,7 +1643,7 @@ def plan_builder(
         if always_include_local_changes is None:
             # default behaviour - if restatements are detected; we operate entirely out of state and ignore local changes
             force_no_diff = restate_models is not None or (
-                backfill_models is not None and not backfill_models
+                backfill_models is not None and not backfill_models and not selected_deletion_fqns
             )
         else:
             force_no_diff = not always_include_local_changes

sqlmesh/core/environment.py

@@ -56,7 +56,8 @@ def _sanitize_name(cls, v: str) -> str:
     @classmethod
     def _validate_boolean_field(cls, v: t.Any, info: ValidationInfo) -> bool:
         if v is None:
-            return info.field_name == "normalize_name"
+            # Pydantic 2.13+ sets field_name to None during model_validate_json()
+            return (info.field_name or "") == "normalize_name"
         return bool(v)
 
     @t.overload

sqlmesh/core/metric/definition.py

@@ -10,7 +10,7 @@
 from sqlmesh.core.node import str_or_exp_to_str
 from sqlmesh.utils import UniqueKeyDict
 from sqlmesh.utils.errors import ConfigError
-from sqlmesh.utils.pydantic import PydanticModel, ValidationInfo, field_validator
+from sqlmesh.utils.pydantic import PydanticModel, ValidationInfo, field_validator, validation_data
 
 MeasureAndDimTables = t.Tuple[str, t.Tuple[str, ...]]
 
@@ -89,7 +89,7 @@ def _string_validator(cls, v: t.Any) -> t.Optional[str]:
     @field_validator("expression", mode="before")
     def _validate_expression(cls, v: t.Any, info: ValidationInfo) -> exp.Expr:
         if isinstance(v, str):
-            dialect = info.data.get("dialect")
+            dialect = validation_data(info).get("dialect")
             return d.parse_one(v, dialect=dialect)
         if isinstance(v, exp.Expr):
             return v

sqlmesh/core/model/common.py

@@ -21,7 +21,13 @@
     prepare_env,
     serialize_env,
 )
-from sqlmesh.utils.pydantic import PydanticModel, ValidationInfo, field_validator, get_dialect
+from sqlmesh.utils.pydantic import (
+    PydanticModel,
+    ValidationInfo,
+    field_validator,
+    get_dialect,
+    validation_data,
+)
 
 if t.TYPE_CHECKING:
     from sqlglot.dialects.dialect import DialectType
@@ -479,7 +485,7 @@ def parse_expression(
     if callable(v):
         return v
 
-    dialect = info.data.get("dialect") if info else ""
+    dialect = validation_data(info).get("dialect") if info else ""
 
     if isinstance(v, list):
         return [
@@ -519,7 +525,7 @@ def parse_properties(
     if v is None:
         return v
 
-    dialect = info.data.get("dialect") if info else ""
+    dialect = validation_data(info).get("dialect") if info else ""
 
     if isinstance(v, str):
         v = d.parse_one(v, dialect=dialect)
@@ -557,8 +563,9 @@ def default_catalog(cls: t.Type, v: t.Any) -> t.Optional[str]:
 
 
 def depends_on(cls: t.Type, v: t.Any, info: ValidationInfo) -> t.Optional[t.Set[str]]:
-    dialect = info.data.get("dialect")
-    default_catalog = info.data.get("default_catalog")
+    data = validation_data(info)
+    dialect = data.get("dialect")
+    default_catalog = data.get("default_catalog")
 
     if isinstance(v, exp.Paren):
         v = v.unnest()

sqlmesh/core/model/meta.py

@@ -44,6 +44,7 @@
     list_of_fields_validator,
     model_validator,
     get_dialect,
+    validation_data,
 )
 
 if t.TYPE_CHECKING:
@@ -135,7 +136,7 @@ def _func_call_validator(cls, v: t.Any, field: t.Any) -> t.Any:
 
     @field_validator("tags", mode="before")
     def _value_or_tuple_validator(cls, v: t.Any, info: ValidationInfo) -> t.Any:
-        return ensure_list(cls._validate_value_or_tuple(v, info.data))
+        return ensure_list(cls._validate_value_or_tuple(v, validation_data(info)))
 
     @classmethod
     def _validate_value_or_tuple(
@@ -164,7 +165,7 @@ def _normalize(value: t.Any) -> t.Any:
     @field_validator("table_format", "storage_format", mode="before")
     def _format_validator(cls, v: t.Any, info: ValidationInfo) -> t.Optional[str]:
         if isinstance(v, exp.Expr) and not (isinstance(v, (exp.Literal, exp.Identifier))):
-            return v.sql(info.data.get("dialect"))
+            return v.sql(validation_data(info).get("dialect"))
         return str_or_exp_to_str(v)
 
     @field_validator("dialect", mode="before")
@@ -192,7 +193,7 @@ def _partition_and_cluster_validator(cls, v: t.Any, info: ValidationInfo) -> t.L
         if (
             isinstance(v, list)
             and all(isinstance(i, str) for i in v)
-            and info.field_name == "partitioned_by_"
+            and (info.field_name or "") == "partitioned_by_"
         ):
             # this branch gets hit when we are deserializing from json because `partitioned_by` is stored as a List[str]
             # however, we should only invoke this if the list contains strings because this validator is also
@@ -205,7 +206,7 @@ def _partition_and_cluster_validator(cls, v: t.Any, info: ValidationInfo) -> t.L
             )
             v = parsed.this.expressions if isinstance(parsed.this, exp.Schema) else v
 
-        expressions = list_of_fields_validator(v, info.data)
+        expressions = list_of_fields_validator(v, validation_data(info))
 
         for expression in expressions:
             num_cols = len(list(expression.find_all(exp.Column)))
@@ -228,7 +229,7 @@ def _columns_validator(
         cls, v: t.Any, info: ValidationInfo
     ) -> t.Optional[t.Dict[str, exp.DataType]]:
         columns_to_types = {}
-        dialect = info.data.get("dialect")
+        dialect = validation_data(info).get("dialect")
 
         if isinstance(v, exp.Schema):
             for column in v.expressions:
@@ -280,7 +281,8 @@ def _columns_validator(
     def _column_descriptions_validator(
         cls, vs: t.Any, info: ValidationInfo
     ) -> t.Optional[t.Dict[str, str]]:
-        dialect = info.data.get("dialect")
+        data = validation_data(info)
+        dialect = data.get("dialect")
 
         if vs is None:
             return None
@@ -302,23 +304,23 @@ def _column_descriptions_validator(
             for k, v in raw_col_descriptions.items()
         }
 
-        columns_to_types = info.data.get("columns_to_types_")
+        columns_to_types = data.get("columns_to_types_")
         if columns_to_types:
             from sqlmesh.core.console import get_console
 
             console = get_console()
             for column_name in list(col_descriptions):
                 if column_name not in columns_to_types:
                     console.log_warning(
-                        f"In model '{info.data['name']}', a description is provided for column '{column_name}' but it is not a column in the model."
+                        f"In model '{data.get('name', '<unknown>')}', a description is provided for column '{column_name}' but it is not a column in the model."
                     )
                     del col_descriptions[column_name]
 
         return col_descriptions
 
     @field_validator("grains", "references", mode="before")
     def _refs_validator(cls, vs: t.Any, info: ValidationInfo) -> t.List[exp.Expr]:
-        dialect = info.data.get("dialect")
+        dialect = validation_data(info).get("dialect")
 
         if isinstance(vs, exp.Paren):
             vs = vs.unnest()

sqlmesh/core/selector.py

@@ -62,7 +62,7 @@ def select_models(
         target_env_name: str,
         fallback_env_name: t.Optional[str] = None,
         ensure_finalized_snapshots: bool = False,
-    ) -> UniqueKeyDict[str, Model]:
+    ) -> t.Tuple[UniqueKeyDict[str, Model], t.Set[str]]:
         """Given a set of selections returns models from the current state with names matching the
         selection while sourcing the remaining models from the target environment.
 
@@ -76,29 +76,11 @@ def select_models(
                 the environment is not finalized.
 
         Returns:
-            A dictionary of models.
+            A tuple of (models dict, set of all matched FQNs including env models).
         """
-        target_env = self._state_reader.get_environment(Environment.sanitize_name(target_env_name))
-        if target_env and target_env.expired:
-            target_env = None
-
-        if not target_env and fallback_env_name:
-            target_env = self._state_reader.get_environment(
-                Environment.sanitize_name(fallback_env_name)
-            )
-
-        env_models: t.Dict[str, Model] = {}
-        if target_env:
-            environment_snapshot_infos = (
-                target_env.snapshots
-                if not ensure_finalized_snapshots
-                else target_env.finalized_or_current_snapshots
-            )
-            env_models = {
-                s.name: s.model
-                for s in self._state_reader.get_snapshots(environment_snapshot_infos).values()
-                if s.is_model
-            }
+        env_models = self._load_env_models(
+            target_env_name, fallback_env_name, ensure_finalized_snapshots
+        )
 
         all_selected_models = self.expand_model_selections(
             model_selections, models={**env_models, **self._models}
@@ -166,7 +148,37 @@ def get_model(fqn: str) -> t.Optional[Model]:
         if needs_update:
             update_model_schemas(dag, models=models, cache_dir=self._cache_dir)
 
-        return models
+        return models, all_selected_models
+
+    def _load_env_models(
+        self,
+        target_env_name: str,
+        fallback_env_name: t.Optional[str] = None,
+        ensure_finalized_snapshots: bool = False,
+    ) -> t.Dict[str, "Model"]:
+        """Loads models from the target environment, falling back to the fallback environment if needed."""
+        target_env = self._state_reader.get_environment(Environment.sanitize_name(target_env_name))
+        if target_env and target_env.expired:
+            target_env = None
+
+        if not target_env and fallback_env_name:
+            target_env = self._state_reader.get_environment(
+                Environment.sanitize_name(fallback_env_name)
+            )
+
+        if not target_env:
+            return {}
+
+        environment_snapshot_infos = (
+            target_env.snapshots
+            if not ensure_finalized_snapshots
+            else target_env.finalized_or_current_snapshots
+        )
+        return {
+            s.name: s.model
+            for s in self._state_reader.get_snapshots(environment_snapshot_infos).values()
+            if s.is_model
+        }
 
     def expand_model_selections(
         self, model_selections: t.Iterable[str], models: t.Optional[t.Dict[str, Node]] = None

sqlmesh/core/state_sync/common.py

@@ -11,7 +11,7 @@
 from pydantic_core.core_schema import ValidationInfo
 from sqlglot import exp
 
-from sqlmesh.utils.pydantic import PydanticModel, field_validator
+from sqlmesh.utils.pydantic import PydanticModel, field_validator, validation_data
 from sqlmesh.core.environment import Environment, EnvironmentStatements, EnvironmentNamingInfo
 from sqlmesh.core.snapshot import (
     Snapshot,
@@ -269,7 +269,7 @@ class PromotionResult(PydanticModel):
     def _validate_removed_environment_naming_info(
         cls, v: t.Optional[EnvironmentNamingInfo], info: ValidationInfo
     ) -> t.Optional[EnvironmentNamingInfo]:
-        if v and not info.data.get("removed"):
+        if v and not validation_data(info).get("removed"):
             raise ValueError("removed_environment_naming_info must be None if removed is empty")
         return v
 

sqlmesh/core/state_sync/db/snapshot.py

@@ -138,7 +138,7 @@ def unpause_snapshots(
         ):
             self.engine_adapter.update_table(
                 self.snapshots_table,
-                {"unpaused_ts": None, "updated_ts": updated_ts},
+                {"unpaused_ts": None},
                 where=where,
             )
 

sqlmesh/core/user.py

@@ -2,7 +2,7 @@
 from enum import Enum
 
 from sqlmesh.core.notification_target import BasicSMTPNotificationTarget, NotificationTarget
-from sqlmesh.utils.pydantic import PydanticModel, ValidationInfo, field_validator
+from sqlmesh.utils.pydantic import PydanticModel, ValidationInfo, field_validator, validation_data
 
 
 class UserRole(str, Enum):
@@ -42,7 +42,7 @@ def validate_notification_targets(
         v: t.List[NotificationTarget],
         info: ValidationInfo,
     ) -> t.List[NotificationTarget]:
-        email = info.data["email"]
+        email = validation_data(info).get("email")
         for target in v:
             if isinstance(target, BasicSMTPNotificationTarget) and target.recipients != {email}:
                 raise ValueError("Recipient emails do not match user email")