Merge branch 'main' into jo/warn_on_variabla_render_failure

Author: georgesittas

pyproject.toml

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

sqlmesh/core/engine_adapter/bigquery.py

@@ -169,17 +169,18 @@ def _df_to_source_queries(
         )
 
         def query_factory() -> Query:
-            if bigframes_pd and isinstance(df, bigframes_pd.DataFrame):
-                df.to_gbq(
+            ordered_df = df[list(source_columns_to_types)]
+            if bigframes_pd and isinstance(ordered_df, bigframes_pd.DataFrame):
+                ordered_df.to_gbq(
                     f"{temp_bq_table.project}.{temp_bq_table.dataset_id}.{temp_bq_table.table_id}",
                     if_exists="replace",
                 )
             elif not self.table_exists(temp_table):
                 # Make mypy happy
-                assert isinstance(df, pd.DataFrame)
+                assert isinstance(ordered_df, pd.DataFrame)
                 self._db_call(self.client.create_table, table=temp_bq_table, exists_ok=False)
                 result = self.__load_pandas_to_table(
-                    temp_bq_table, df, source_columns_to_types, replace=False
+                    temp_bq_table, ordered_df, source_columns_to_types, replace=False
                 )
                 if result.errors:
                     raise SQLMeshError(result.errors)

sqlmesh/core/engine_adapter/snowflake.py

@@ -378,6 +378,8 @@ def query_factory() -> Query:
             elif isinstance(df, pd.DataFrame):
                 from snowflake.connector.pandas_tools import write_pandas
 
+                ordered_df = df[list(source_columns_to_types)]
+
                 # Workaround for https://github.com/snowflakedb/snowflake-connector-python/issues/1034
                 # The above issue has already been fixed upstream, but we keep the following
                 # line anyway in order to support a wider range of Snowflake versions.
@@ -388,16 +390,16 @@ def query_factory() -> Query:
 
                 # See: https://stackoverflow.com/a/75627721
                 for column, kind in source_columns_to_types.items():
-                    if is_datetime64_any_dtype(df.dtypes[column]):
+                    if is_datetime64_any_dtype(ordered_df.dtypes[column]):
                         if kind.is_type("date"):  # type: ignore
-                            df[column] = pd.to_datetime(df[column]).dt.date  # type: ignore
-                        elif getattr(df.dtypes[column], "tz", None) is not None:  # type: ignore
-                            df[column] = pd.to_datetime(df[column]).dt.strftime(
+                            ordered_df[column] = pd.to_datetime(ordered_df[column]).dt.date  # type: ignore
+                        elif getattr(ordered_df.dtypes[column], "tz", None) is not None:  # type: ignore
+                            ordered_df[column] = pd.to_datetime(ordered_df[column]).dt.strftime(
                                 "%Y-%m-%d %H:%M:%S.%f%z"
                             )  # type: ignore
                         # https://github.com/snowflakedb/snowflake-connector-python/issues/1677
                         else:  # type: ignore
-                            df[column] = pd.to_datetime(df[column]).dt.strftime(
+                            ordered_df[column] = pd.to_datetime(ordered_df[column]).dt.strftime(
                                 "%Y-%m-%d %H:%M:%S.%f"
                             )  # type: ignore
 
@@ -407,7 +409,7 @@ def query_factory() -> Query:
 
                 write_pandas(
                     self._connection_pool.get(),
-                    df,
+                    ordered_df,
                     temp_table.name,
                     schema=temp_table.db or None,
                     database=database.sql(dialect=self.dialect) if database else None,

sqlmesh/core/plan/common.py

@@ -16,7 +16,11 @@ 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 and not new.is_metadata:
+    if new.is_seed and not (
+        new.is_metadata
+        and new.previous_version
+        and new.previous_version.snapshot_id(new.name) == old.snapshot_id
+    ):
         # 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

sqlmesh/core/plan/stages.py

@@ -268,7 +268,7 @@ def build(self, plan: EvaluatablePlan) -> t.List[PlanStage]:
             before_promote_snapshots = {
                 s.snapshot_id
                 for s in snapshots.values()
-                if deployability_index.is_representative(s)
+                if (deployability_index.is_representative(s) or s.is_seed)
                 and plan.is_selected_for_backfill(s.name)
             }
             after_promote_snapshots = all_selected_for_backfill_snapshots - before_promote_snapshots

sqlmesh/core/snapshot/evaluator.py

@@ -1141,10 +1141,10 @@ def _migrate_target_table(
     ) -> None:
         adapter = self.get_adapter(snapshot.model.gateway)
 
-        target_table = exp.to_table(target_table_name)
-        target_table.this.set("this", f"{target_table.name}_schema_tmp")
+        tmp_table = exp.to_table(target_table_name)
+        tmp_table.this.set("this", f"{tmp_table.name}_schema_tmp")
+        tmp_table_name = tmp_table.sql()
 
-        tmp_table_name = target_table.sql()
         if snapshot.is_materialized:
             self._execute_create(
                 snapshot=snapshot,
@@ -2185,6 +2185,18 @@ def create(
             self.adapter.drop_table(table_name)
             raise
 
+    def migrate(
+        self,
+        target_table_name: str,
+        source_table_name: str,
+        snapshot: Snapshot,
+        *,
+        ignore_destructive: bool,
+        ignore_additive: bool,
+        **kwargs: t.Any,
+    ) -> None:
+        raise NotImplementedError("Seeds do not support migrations.")
+
     def insert(
         self,
         table_name: str,

sqlmesh/utils/jinja.py

@@ -206,6 +206,20 @@ def extract_macro_references_and_variables(
     return macro_references, variables
 
 
+def sort_dict_recursive(
+    item: t.Dict[str, t.Any],
+) -> t.Dict[str, t.Any]:
+    sorted_dict: t.Dict[str, t.Any] = {}
+    for k, v in sorted(item.items()):
+        if isinstance(v, list):
+            sorted_dict[k] = sorted(v)
+        elif isinstance(v, dict):
+            sorted_dict[k] = sort_dict_recursive(v)
+        else:
+            sorted_dict[k] = v
+    return sorted_dict
+
+
 JinjaGlobalAttribute = t.Union[str, int, float, bool, AttributeDict]
 
 
@@ -440,7 +454,7 @@ def to_expressions(self) -> t.List[Expression]:
                 d.PythonCode(
                     expressions=[
                         f"{k} = '{v}'" if isinstance(v, str) else f"{k} = {v}"
-                        for k, v in sorted(filtered_objs.items())
+                        for k, v in sort_dict_recursive(filtered_objs).items()
                     ]
                 )
             )

tests/core/engine_adapter/test_bigquery.py

@@ -487,7 +487,13 @@ def temp_table_exists(table: exp.Table) -> bool:
     retry_resp_call.errors = None
     retry_mock.return_value = retry_resp
     db_call_mock.return_value = AttributeDict({"errors": None})
-    df = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
+    df = pd.DataFrame(
+        {
+            "id": [1, 2, 3],
+            "ts": ["2025-01-01 00:00:00", "2025-01-01 00:00:00", "2025-01-01 00:00:00"],
+            "val": [7, 8, 9],
+        }
+    )
     adapter.merge(
         target_table="target",
         source_table=df,

tests/core/test_integration.py

@@ -3232,6 +3232,62 @@ def test_virtual_environment_mode_dev_only_model_change_standalone_audit(
     context.apply(plan)
 
 
+@time_machine.travel("2023-01-08 15:00:00 UTC")
+def test_virtual_environment_mode_dev_only_seed_model_change_schema(
+    init_and_plan_context: t.Callable,
+):
+    context, plan = init_and_plan_context(
+        "examples/sushi", config="test_config_virtual_environment_mode_dev_only"
+    )
+    context.apply(plan)
+
+    new_csv = []
+    with open(context.path / "seeds" / "waiter_names.csv", "r") as fd:
+        is_header = True
+        for idx, line in enumerate(fd):
+            line = line.strip()
+            if not line:
+                continue
+            if is_header:
+                new_csv.append(line + ",new_column")
+                is_header = False
+            else:
+                new_csv.append(line + f",v{idx}")
+
+    with open(context.path / "seeds" / "waiter_names.csv", "w") as fd:
+        fd.write("\n".join(new_csv))
+
+    context.load()
+
+    downstream_model = context.get_model("sushi.waiter_as_customer_by_day")
+    downstream_model_kind = downstream_model.kind.dict()
+    downstream_model_kwargs = {
+        **downstream_model.dict(),
+        "kind": {
+            **downstream_model_kind,
+            "on_destructive_change": "allow",
+        },
+        "audits": [],
+        # Use the new column
+        "query": "SELECT '2023-01-07' AS event_date, new_column AS new_column FROM sushi.waiter_names",
+    }
+    context.upsert_model(SqlModel.parse_obj(downstream_model_kwargs))
+
+    context.plan("dev", auto_apply=True, no_prompts=True, skip_tests=True, enable_preview=True)
+
+    assert (
+        context.engine_adapter.fetchone(
+            "SELECT COUNT(*) FROM sushi__dev.waiter_as_customer_by_day"
+        )[0]
+        == len(new_csv) - 1
+    )
+
+    # Deploy to prod
+    context.clear_caches()
+    context.plan("prod", auto_apply=True, no_prompts=True, skip_tests=True)
+    assert "new_column" in context.engine_adapter.columns("sushi.waiter_as_customer_by_day")
+
+
 @time_machine.travel("2023-01-08 15:00:00 UTC")
 def test_restatement_plan_ignores_changes(init_and_plan_context: t.Callable):
     context, plan = init_and_plan_context("examples/sushi")

tests/core/test_plan.py

@@ -1214,6 +1214,7 @@ def test_seed_model_metadata_change_no_missing_intervals(
             description="foo",
         )
     )
+    snapshot_a_metadata_updated.previous_versions = snapshot_a.all_versions
     assert snapshot_a_metadata_updated.version is None
     assert snapshot_a_metadata_updated.change_category is None