Merge branch 'main' into fix/sqlmesh-stale-cache-file-not-found

Author: Pavan-Rana

pyproject.toml

@@ -24,7 +24,7 @@ dependencies = [
     "requests",
     "rich[jupyter]",
     "ruamel.yaml",
-    "sqlglot~=30.2.1",
+    "sqlglot~=30.4.2",
     "tenacity",
     "time-machine",
     "json-stream"

sqlmesh/core/config/connection.py

@@ -34,6 +34,7 @@
     ValidationInfo,
     field_validator,
     model_validator,
+    validation_data,
     validation_error_message,
     get_concrete_types_from_typehint,
 )
@@ -1081,7 +1082,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."
             )
@@ -1093,7 +1094,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/engine_adapter/base.py

@@ -2920,7 +2920,11 @@ def _replace_by_key(
             target_columns_to_types = self.columns(target_table)
 
         temp_table = self._get_temp_table(target_table)
-        key_exp = exp.func("CONCAT_WS", "'__SQLMESH_DELIM__'", *key) if len(key) > 1 else key[0]
+        key_exp = (
+            exp.func("CONCAT_WS", "'__SQLMESH_DELIM__'", *key, dialect=self.dialect)
+            if len(key) > 1
+            else key[0]
+        )
         column_names = list(target_columns_to_types or [])
 
         with self.transaction():

sqlmesh/core/engine_adapter/clickhouse.py

@@ -423,7 +423,11 @@ def _replace_by_key(
             source_columns=source_columns,
         )
 
-        key_exp = exp.func("CONCAT_WS", "'__SQLMESH_DELIM__'", *key) if len(key) > 1 else key[0]
+        key_exp = (
+            exp.func("CONCAT_WS", "'__SQLMESH_DELIM__'", *key, dialect=self.dialect)
+            if len(key) > 1
+            else key[0]
+        )
 
         self._insert_overwrite_by_condition(
             target_table,

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/definition.py

@@ -1158,11 +1158,7 @@ def _audit_metadata_hash_values(self) -> t.List[str]:
 
         for audit_name, audit_args in sorted(self.audits, key=lambda a: a[0]):
             metadata.append(audit_name)
-            if audit_name in BUILT_IN_AUDITS:
-                for arg_name, arg_value in audit_args.items():
-                    metadata.append(arg_name)
-                    metadata.append(gen(arg_value))
-            else:
+            if audit_name not in BUILT_IN_AUDITS:
                 audit = self.audit_definitions[audit_name]
                 metadata.extend(
                     [
@@ -1172,6 +1168,9 @@ def _audit_metadata_hash_values(self) -> t.List[str]:
                         str(audit.blocking),
                     ]
                 )
+            for arg_name, arg_value in audit_args.items():
+                metadata.append(arg_name)
+                metadata.append(gen(arg_value))
 
         return metadata
 

sqlmesh/core/model/kind.py

@@ -786,7 +786,7 @@ def data_hash_values(self) -> t.List[t.Optional[str]]:
             gen(self.valid_to_name),
             str(self.invalidate_hard_deletes),
             self.time_data_type.sql(self.dialect),
-            gen(self.batch_size) if self.batch_size is not None else None,
+            str(self.batch_size) if self.batch_size is not None else None,
         ]
 
     @property