Merge branch 'main' into indirect_change_to_materialized_view_is_breaking

Author: xardasos

pyproject.toml

@@ -24,7 +24,7 @@ dependencies = [
     "requests",
     "rich[jupyter]",
     "ruamel.yaml",
-    "sqlglot[rs]~=27.14.0",
+    "sqlglot[rs]~=27.16.3",
     "tenacity",
     "time-machine",
     "json-stream"
@@ -56,7 +56,7 @@ dev = [
     "agate",
     "beautifulsoup4",
     "clickhouse-connect",
-    "cryptography",
+    "cryptography<46.0.0",
     "databricks-sql-connector",
     "dbt-bigquery",
     "dbt-core",
@@ -119,7 +119,7 @@ postgres = ["psycopg2"]
 redshift = ["redshift_connector"]
 slack = ["slack_sdk"]
 snowflake = [
-    "cryptography",
+    "cryptography<46.0.0",
     "snowflake-connector-python[pandas,secure-local-storage]",
     "snowflake-snowpark-python",
 ]
@@ -135,7 +135,7 @@ lsp = [
     # Duplicate of web
     "fastapi==0.115.5",
     "watchfiles>=0.19.0",
-    "uvicorn[standard]==0.22.0",
+    # "uvicorn[standard]==0.22.0",
     "sse-starlette>=0.2.2",
     "pyarrow",
     # For lsp

sqlmesh/core/engine_adapter/base.py

@@ -1044,6 +1044,7 @@ def clone_table(
         target_table_name: TableName,
         source_table_name: TableName,
         replace: bool = False,
+        exists: bool = True,
         clone_kwargs: t.Optional[t.Dict[str, t.Any]] = None,
         **kwargs: t.Any,
     ) -> None:
@@ -1053,6 +1054,7 @@ def clone_table(
             target_table_name: The name of the table that should be created.
             source_table_name: The name of the source table that should be cloned.
             replace: Whether or not to replace an existing table.
+            exists: Indicates whether to include the IF NOT EXISTS check.
         """
         if not self.SUPPORTS_CLONING:
             raise NotImplementedError(f"Engine does not support cloning: {type(self)}")
@@ -1063,6 +1065,7 @@ def clone_table(
                 this=exp.to_table(target_table_name),
                 kind="TABLE",
                 replace=replace,
+                exists=exists,
                 clone=exp.Clone(
                     this=exp.to_table(source_table_name),
                     **(clone_kwargs or {}),

sqlmesh/core/engine_adapter/databricks.py

@@ -299,6 +299,7 @@ def clone_table(
         target_table_name: TableName,
         source_table_name: TableName,
         replace: bool = False,
+        exists: bool = True,
         clone_kwargs: t.Optional[t.Dict[str, t.Any]] = None,
         **kwargs: t.Any,
     ) -> None:

sqlmesh/core/engine_adapter/snowflake.py

@@ -610,6 +610,7 @@ def clone_table(
         target_table_name: TableName,
         source_table_name: TableName,
         replace: bool = False,
+        exists: bool = True,
         clone_kwargs: t.Optional[t.Dict[str, t.Any]] = None,
         **kwargs: t.Any,
     ) -> None:

sqlmesh/core/model/definition.py

@@ -2866,6 +2866,13 @@ def render_field_value(value: t.Any) -> t.Any:
             for key, value in field_value.items():
                 if key in RUNTIME_RENDERED_MODEL_FIELDS:
                     rendered_dict[key] = parse_strings_with_macro_refs(value, dialect)
+                elif (
+                    # don't parse kind auto_restatement_cron="@..." kwargs (e.g. @daily) into MacroVar
+                    key == "auto_restatement_cron"
+                    and isinstance(value, str)
+                    and value.lower() in CRON_SHORTCUTS
+                ):
+                    rendered_dict[key] = value
                 elif (rendered := render_field_value(value)) is not None:
                     rendered_dict[key] = rendered
 

sqlmesh/core/plan/common.py

@@ -16,8 +16,9 @@ def should_force_rebuild(old: Snapshot, new: Snapshot) -> bool:
     if new.is_view and new.is_indirect_non_breaking and not new.is_forward_only:
         # View models always need to be rebuilt to reflect updated upstream dependencies
         return True
-    if new.is_seed:
+    if new.is_seed and not new.is_metadata:
         # Seed models always need to be rebuilt to reflect changes in the seed file
+        # Unless only their metadata has been updated (eg description added) and the seed file has not been touched
         return True
     return is_breaking_kind_change(old, new)
 

sqlmesh/core/renderer.py

@@ -6,7 +6,7 @@
 from functools import partial
 from pathlib import Path
 
-from sqlglot import exp, parse
+from sqlglot import exp, Dialect
 from sqlglot.errors import SqlglotError
 from sqlglot.helper import ensure_list
 from sqlglot.optimizer.annotate_types import annotate_types
@@ -249,15 +249,24 @@ def _resolve_table(table: str | exp.Table) -> str:
                 ) from ex
 
             if rendered_expression.strip():
-                try:
-                    expressions = [e for e in parse(rendered_expression, read=self._dialect) if e]
-
-                    if not expressions:
-                        raise ConfigError(f"Failed to parse an expression:\n{self._expression}")
-                except Exception as ex:
-                    raise ConfigError(
-                        f"Could not parse the rendered jinja at '{self._path}'.\n{ex}"
-                    ) from ex
+                # ensure there is actual SQL and not just comments and non-SQL jinja
+                dialect = Dialect.get_or_raise(self._dialect)
+                tokens = dialect.tokenize(rendered_expression)
+
+                if tokens:
+                    try:
+                        expressions = [
+                            e for e in dialect.parser().parse(tokens, rendered_expression) if e
+                        ]
+
+                        if not expressions:
+                            raise ConfigError(
+                                f"Failed to parse an expression:\n{rendered_expression}"
+                            )
+                    except Exception as ex:
+                        raise ConfigError(
+                            f"Could not parse the rendered jinja at '{self._path}'.\n{ex}"
+                        ) from ex
 
         if this_model:
             render_kwargs["this_model"] = this_model

sqlmesh/core/snapshot/definition.py

@@ -1480,7 +1480,7 @@ def expiration_ts(self) -> int:
     @property
     def supports_schema_migration_in_prod(self) -> bool:
         """Returns whether or not this snapshot supports schema migration when deployed to production."""
-        return self.is_paused and self.is_model and not self.is_symbolic
+        return self.is_paused and self.is_model and not self.is_symbolic and not self.is_seed
 
     @property
     def requires_schema_migration_in_prod(self) -> bool:

sqlmesh/core/snapshot/evaluator.py

@@ -32,6 +32,7 @@
 
 from sqlglot import exp, select
 from sqlglot.executor import execute
+from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_not_exception_type
 
 from sqlmesh.core import constants as c
 from sqlmesh.core import dialect as d
@@ -76,6 +77,7 @@
 from sqlmesh.utils.errors import (
     ConfigError,
     DestructiveChangeError,
+    MigrationNotSupportedError,
     SQLMeshError,
     format_destructive_change_msg,
     format_additive_change_msg,
@@ -492,7 +494,7 @@ def migrate(
         with self.concurrent_context():
             # Only migrate snapshots for which there's an existing data object
             concurrent_apply_to_snapshots(
-                snapshots_by_name.values(),
+                target_snapshots,
                 lambda s: self._migrate_snapshot(
                     s,
                     snapshots_by_name,
@@ -761,7 +763,7 @@ def _evaluate_snapshot(
                         snapshots=snapshots,
                         deployability_index=deployability_index,
                         render_kwargs=create_render_kwargs,
-                        rendered_physical_properties=rendered_physical_properties,
+                        rendered_physical_properties=rendered_physical_properties.copy(),
                         allow_destructive_snapshots=allow_destructive_snapshots,
                         allow_additive_snapshots=allow_additive_snapshots,
                     )
@@ -774,7 +776,7 @@ def _evaluate_snapshot(
                         is_table_deployable=is_snapshot_deployable,
                         deployability_index=deployability_index,
                         create_render_kwargs=create_render_kwargs,
-                        rendered_physical_properties=rendered_physical_properties,
+                        rendered_physical_properties=rendered_physical_properties.copy(),
                         dry_run=False,
                         run_pre_post_statements=False,
                     )
@@ -865,6 +867,7 @@ def create_snapshot(
                     rendered_physical_properties=rendered_physical_properties,
                     allow_destructive_snapshots=allow_destructive_snapshots,
                     allow_additive_snapshots=allow_additive_snapshots,
+                    run_pre_post_statements=True,
                 )
             else:
                 is_table_deployable = deployability_index.is_deployable(snapshot)
@@ -1024,6 +1027,7 @@ def _clone_snapshot_in_dev(
         rendered_physical_properties: t.Dict[str, exp.Expression],
         allow_destructive_snapshots: t.Set[str],
         allow_additive_snapshots: t.Set[str],
+        run_pre_post_statements: bool = False,
     ) -> None:
         adapter = self.get_adapter(snapshot.model.gateway)
 
@@ -1035,7 +1039,6 @@ def _clone_snapshot_in_dev(
             adapter.clone_table(
                 target_table_name,
                 snapshot.table_name(),
-                replace=True,
                 rendered_physical_properties=rendered_physical_properties,
             )
             self._migrate_target_table(
@@ -1047,6 +1050,7 @@ def _clone_snapshot_in_dev(
                 rendered_physical_properties=rendered_physical_properties,
                 allow_destructive_snapshots=allow_destructive_snapshots,
                 allow_additive_snapshots=allow_additive_snapshots,
+                run_pre_post_statements=run_pre_post_statements,
             )
         except Exception:
             adapter.drop_table(target_table_name)
@@ -1111,6 +1115,15 @@ def _migrate_snapshot(
                     dry_run=True,
                 )
 
+    # Retry in case when the table is migrated concurrently from another plan application
+    @retry(
+        reraise=True,
+        stop=stop_after_attempt(5),
+        wait=wait_exponential(min=1, max=16),
+        retry=retry_if_not_exception_type(
+            (DestructiveChangeError, AdditiveChangeError, MigrationNotSupportedError)
+        ),
+    )
     def _migrate_target_table(
         self,
         target_table_name: str,
@@ -1440,8 +1453,9 @@ def _can_clone(self, snapshot: Snapshot, deployability_index: DeployabilityIndex
             and adapter.SUPPORTS_CLONING
             # managed models cannot have their schema mutated because theyre based on queries, so clone + alter wont work
             and not snapshot.is_managed
-            # If the deployable table is missing we can't clone it
             and not deployability_index.is_deployable(snapshot)
+            # If the deployable table is missing we can't clone it
+            and adapter.table_exists(snapshot.table_name())
         )
 
     def _get_data_objects(
@@ -2671,7 +2685,7 @@ def migrate(
         )
         if len(potential_alter_operations) > 0:
             # this can happen if a user changes a managed model and deliberately overrides a plan to be forward only, eg `sqlmesh plan --forward-only`
-            raise SQLMeshError(
+            raise MigrationNotSupportedError(
                 f"The schema of the managed model '{target_table_name}' cannot be updated in a forward-only fashion."
             )
 

sqlmesh/core/test/definition.py

@@ -644,16 +644,16 @@ def _create_df(
             return self._execute(query)
 
         rows = values["rows"]
+        columns_str: t.Optional[t.List[str]] = None
         if columns:
+            columns_str = [str(c) for c in columns]
             referenced_columns = list(dict.fromkeys(col for row in rows for col in row))
             _raise_if_unexpected_columns(columns, referenced_columns)
 
             if partial:
-                columns = referenced_columns
+                columns_str = [c for c in columns_str if c in referenced_columns]
 
-        return pd.DataFrame.from_records(
-            rows, columns=[str(c) for c in columns] if columns else None
-        )
+        return pd.DataFrame.from_records(rows, columns=columns_str)
 
     def _add_missing_columns(
         self, query: exp.Query, all_columns: t.Optional[t.Collection[str]] = None