Merge branch 'main' into trey/comment-jinja-model

georgesittas · web-flow · commit a7d353b810bf · 2025-09-18T16:21:53.000+03:00
diff --git a/pyproject.toml b/pyproject.toml
@@ -24,7 +24,7 @@ dependencies = [
     "requests",
     "rich[jupyter]",
     "ruamel.yaml",
-    "sqlglot[rs]~=27.14.0",
+    "sqlglot[rs]~=27.15.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
diff --git a/sqlmesh/core/engine_adapter/base.py b/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 {}),
diff --git a/sqlmesh/core/engine_adapter/databricks.py b/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:
diff --git a/sqlmesh/core/engine_adapter/snowflake.py b/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:
diff --git a/sqlmesh/core/plan/common.py b/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)
 
diff --git a/sqlmesh/core/snapshot/evaluator.py b/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,
@@ -1035,7 +1037,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(
@@ -1111,6 +1112,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 +1450,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 +2682,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."
             )
 
diff --git a/sqlmesh/utils/errors.py b/sqlmesh/utils/errors.py
@@ -151,6 +151,10 @@ class AdditiveChangeError(SQLMeshError):
     pass
 
 
+class MigrationNotSupportedError(SQLMeshError):
+    pass
+
+
 class NotificationTargetError(SQLMeshError):
     pass
 
diff --git a/tests/core/engine_adapter/integration/test_integration.py b/tests/core/engine_adapter/integration/test_integration.py
@@ -22,6 +22,7 @@
 
 from sqlmesh import Config, Context
 from sqlmesh.cli.project_init import init_example_project
+from sqlmesh.core.config.common import VirtualEnvironmentMode
 from sqlmesh.core.config.connection import ConnectionConfig
 import sqlmesh.core.dialect as d
 from sqlmesh.core.environment import EnvironmentSuffixTarget
@@ -1938,7 +1939,12 @@ def test_transaction(ctx: TestContext):
     ctx.compare_with_current(table, input_data)
 
 
-def test_sushi(ctx: TestContext, tmp_path: pathlib.Path):
+@pytest.mark.parametrize(
+    "virtual_environment_mode", [VirtualEnvironmentMode.FULL, VirtualEnvironmentMode.DEV_ONLY]
+)
+def test_sushi(
+    ctx: TestContext, tmp_path: pathlib.Path, virtual_environment_mode: VirtualEnvironmentMode
+):
     if ctx.mark == "athena_hive":
         pytest.skip(
             "Sushi end-to-end tests only need to run once for Athena because sushi needs a hybrid of both Hive and Iceberg"
@@ -1984,6 +1990,7 @@ def _mutate_config(gateway: str, config: Config) -> None:
             ).sql(dialect=config.model_defaults.dialect)
             for e in before_all
         ]
+        config.virtual_environment_mode = virtual_environment_mode
 
     context = ctx.create_context(_mutate_config, path=tmp_path, ephemeral_state_connection=False)
 
diff --git a/tests/core/engine_adapter/test_base.py b/tests/core/engine_adapter/test_base.py
@@ -3347,7 +3347,7 @@ def test_clone_table(make_mocked_engine_adapter: t.Callable):
     adapter.clone_table("target_table", "source_table")
 
     adapter.cursor.execute.assert_called_once_with(
-        "CREATE TABLE `target_table` CLONE `source_table`"
+        "CREATE TABLE IF NOT EXISTS `target_table` CLONE `source_table`"
     )
 
 
diff --git a/tests/core/engine_adapter/test_databricks.py b/tests/core/engine_adapter/test_databricks.py
@@ -106,7 +106,7 @@ def test_clone_table(mocker: MockFixture, make_mocked_engine_adapter: t.Callable
     adapter = make_mocked_engine_adapter(DatabricksEngineAdapter, default_catalog="test_catalog")
     adapter.clone_table("target_table", "source_table")
     adapter.cursor.execute.assert_called_once_with(
-        "CREATE TABLE `target_table` SHALLOW CLONE `source_table`"
+        "CREATE TABLE IF NOT EXISTS `target_table` SHALLOW CLONE `source_table`"
     )
 
 
diff --git a/tests/core/engine_adapter/test_snowflake.py b/tests/core/engine_adapter/test_snowflake.py
@@ -688,7 +688,7 @@ def test_clone_table(mocker: MockerFixture, make_mocked_engine_adapter: t.Callab
     adapter = make_mocked_engine_adapter(SnowflakeEngineAdapter, default_catalog="test_catalog")
     adapter.clone_table("target_table", "source_table")
     adapter.cursor.execute.assert_called_once_with(
-        'CREATE TABLE "target_table" CLONE "source_table"'
+        'CREATE TABLE IF NOT EXISTS "target_table" CLONE "source_table"'
     )
 
     # Validate with transient type we create the clone table accordingly
@@ -700,7 +700,7 @@ def test_clone_table(mocker: MockerFixture, make_mocked_engine_adapter: t.Callab
         "target_table", "source_table", rendered_physical_properties=rendered_physical_properties
     )
     adapter.cursor.execute.assert_called_once_with(
-        'CREATE TRANSIENT TABLE "target_table" CLONE "source_table"'
+        'CREATE TRANSIENT TABLE IF NOT EXISTS "target_table" CLONE "source_table"'
     )
 
     # Validate other engine adapters would work as usual even when we pass the properties
@@ -710,7 +710,7 @@ def test_clone_table(mocker: MockerFixture, make_mocked_engine_adapter: t.Callab
         "target_table", "source_table", rendered_physical_properties=rendered_physical_properties
     )
     adapter.cursor.execute.assert_called_once_with(
-        'CREATE TABLE "target_table" CLONE "source_table"'
+        'CREATE TABLE IF NOT EXISTS "target_table" CLONE "source_table"'
     )
 
 
diff --git a/tests/core/test_integration.py b/tests/core/test_integration.py
@@ -10662,3 +10662,121 @@ def _run_restatement_plan(tmp_path: Path, config: Config, q: queue.Queue):
     assert len(model_a.intervals)
 
     set_console(orig_console)
+
+
+def test_seed_model_metadata_update_does_not_trigger_backfill(tmp_path: Path):
+    """
+    Scenario:
+        - Create a seed model; perform initial population
+        - Modify the model with a metadata-only change and trigger a plan
+
+    Outcome:
+        - The seed model is modified (metadata-only) but this should NOT trigger backfill
+        - There should be no missing_intervals on the plan to backfill
+    """
+
+    models_path = tmp_path / "models"
+    seeds_path = tmp_path / "seeds"
+    models_path.mkdir()
+    seeds_path.mkdir()
+
+    seed_model_path = models_path / "seed.sql"
+    seed_path = seeds_path / "seed_data.csv"
+
+    seed_path.write_text("\n".join(["id,name", "1,test"]))
+
+    seed_model_path.write_text("""
+    MODEL (
+        name test.source_data,
+        kind SEED (
+            path '../seeds/seed_data.csv'
+        )
+    );    
+    """)
+
+    config = Config(
+        gateways={"": GatewayConfig(connection=DuckDBConnectionConfig())},
+        model_defaults=ModelDefaultsConfig(dialect="duckdb", start="2024-01-01"),
+    )
+    ctx = Context(paths=tmp_path, config=config)
+
+    plan = ctx.plan(auto_apply=True)
+
+    original_seed_snapshot = ctx.snapshots['"memory"."test"."source_data"']
+    assert plan.directly_modified == {original_seed_snapshot.snapshot_id}
+    assert plan.metadata_updated == set()
+    assert plan.missing_intervals
+
+    # prove data loaded
+    assert ctx.engine_adapter.fetchall("select id, name from memory.test.source_data") == [
+        (1, "test")
+    ]
+
+    # prove no diff
+    ctx.load()
+    plan = ctx.plan(auto_apply=True)
+    assert not plan.has_changes
+    assert not plan.missing_intervals
+
+    # make metadata-only change
+    seed_model_path.write_text("""
+    MODEL (
+        name test.source_data,
+        kind SEED (
+            path '../seeds/seed_data.csv'
+        ),
+        description 'updated by test'
+    );    
+    """)
+
+    ctx.load()
+    plan = ctx.plan(auto_apply=True)
+    assert plan.has_changes
+
+    new_seed_snapshot = ctx.snapshots['"memory"."test"."source_data"']
+    assert (
+        new_seed_snapshot.version == original_seed_snapshot.version
+    )  # should be using the same physical table
+    assert (
+        new_seed_snapshot.snapshot_id != original_seed_snapshot.snapshot_id
+    )  # but still be different due to the metadata change
+    assert plan.directly_modified == set()
+    assert plan.metadata_updated == {new_seed_snapshot.snapshot_id}
+
+    # there should be no missing intervals to backfill since all we did is update a description
+    assert not plan.missing_intervals
+
+    # there should still be no diff or missing intervals in 3 days time
+    assert new_seed_snapshot.model.interval_unit.is_day
+    with time_machine.travel(timedelta(days=3)):
+        ctx.clear_caches()
+        ctx.load()
+        plan = ctx.plan(auto_apply=True)
+        assert not plan.has_changes
+        assert not plan.missing_intervals
+
+    # change seed data
+    seed_path.write_text("\n".join(["id,name", "1,test", "2,updated"]))
+
+    # new plan - NOW we should backfill because data changed
+    ctx.load()
+    plan = ctx.plan(auto_apply=True)
+    assert plan.has_changes
+
+    updated_seed_snapshot = ctx.snapshots['"memory"."test"."source_data"']
+
+    assert (
+        updated_seed_snapshot.snapshot_id
+        != new_seed_snapshot.snapshot_id
+        != original_seed_snapshot.snapshot_id
+    )
+    assert not updated_seed_snapshot.forward_only
+    assert plan.directly_modified == {updated_seed_snapshot.snapshot_id}
+    assert plan.metadata_updated == set()
+    assert plan.missing_intervals
+
+    # prove backfilled data loaded
+    assert ctx.engine_adapter.fetchall("select id, name from memory.test.source_data") == [
+        (1, "test"),
+        (2, "updated"),
+    ]
diff --git a/tests/core/test_plan.py b/tests/core/test_plan.py
diff --git a/tests/core/test_snapshot_evaluator.py b/tests/core/test_snapshot_evaluator.py

Original file line number	Diff line number	Diff line change
`@@ -3347,7 +3347,7 @@ def test_clone_table(make_mocked_engine_adapter: t.Callable):`
`3347`	`3347`	`adapter.clone_table("target_table", "source_table")`
`3348`	`3348`
`3349`	`3349`	`adapter.cursor.execute.assert_called_once_with(`
`3350`		- "CREATE TABLE `target_table` CLONE `source_table`"
	`3350`	+ "CREATE TABLE IF NOT EXISTS `target_table` CLONE `source_table`"
`3351`	`3351`	`)`
`3352`	`3352`
`3353`	`3353`
Original file line number	Diff line number	Diff line change
`@@ -106,7 +106,7 @@ def test_clone_table(mocker: MockFixture, make_mocked_engine_adapter: t.Callable`
`106`	`106`	`adapter = make_mocked_engine_adapter(DatabricksEngineAdapter, default_catalog="test_catalog")`
`107`	`107`	`adapter.clone_table("target_table", "source_table")`
`108`	`108`	`adapter.cursor.execute.assert_called_once_with(`
`109`		- "CREATE TABLE `target_table` SHALLOW CLONE `source_table`"
	`109`	+ "CREATE TABLE IF NOT EXISTS `target_table` SHALLOW CLONE `source_table`"
`110`	`110`	`)`
`111`	`111`
`112`	`112`