fix(snowflake): omit initialization warehouse from preview tables

Signed-off-by: mday-io <mdaytn@gmail.com>

Author: mday-io

sqlmesh/core/engine_adapter/snowflake.py

@@ -343,7 +343,13 @@ def _build_table_properties_exp(
             # if we are creating a non-dynamic table; remove any properties that are only valid for dynamic tables
             # this is necessary because we create "normal" tables from the same managed model definition for dev previews and the "normal" tables dont support these parameters
             if "DYNAMIC" not in (table_kind or "").upper():
-                for prop in {"WAREHOUSE", "TARGET_LAG", "REFRESH_MODE", "INITIALIZE"}:
+                for prop in {
+                    "WAREHOUSE",
+                    "TARGET_LAG",
+                    "REFRESH_MODE",
+                    "INITIALIZE",
+                    "INITIALIZATION_WAREHOUSE",
+                }:
                     table_properties.pop(prop, None)
 
             table_type = self._pop_creatable_type_from_properties(table_properties)

tests/core/engine_adapter/test_snowflake.py

@@ -18,6 +18,8 @@
 from sqlmesh.utils import optional_import
 from tests.core.engine_adapter import to_sql_calls
 from sqlmesh.core.model.kind import ViewKind
+from sqlmesh.core.snapshot import DeployabilityIndex
+from sqlmesh.core.snapshot.evaluator import EngineManagedStrategy
 
 pytestmark = [pytest.mark.engine, pytest.mark.snowflake]
 
@@ -609,6 +611,85 @@ def test_ctas_skips_dynamic_table_properties(make_mocked_engine_adapter: t.Calla
     ]
 
 
+@pytest.mark.parametrize(
+    "operation", ["create_annotated", "create_ctas", "insert", "create_managed"]
+)
+def test_managed_preview_dynamic_table_properties(
+    make_mocked_engine_adapter: t.Callable, mocker: MockerFixture, operation: str
+):
+    adapter = make_mocked_engine_adapter(SnowflakeEngineAdapter)
+    model = load_sql_based_model(
+        d.parse("""
+        MODEL (
+            name test_schema.test_model,
+            dialect snowflake,
+            kind MANAGED,
+            physical_properties (
+                warehouse = 'SMALL',
+                target_lag = '10 minutes',
+                refresh_mode = 'AUTO',
+                initialize = 'ON_CREATE',
+                initialization_warehouse = 'INITIAL_WH',
+                data_retention_time_in_days = 1,
+                max_data_extension_time_in_days = 2
+            )
+        );
+        SELECT a FROM source_table;
+    """)
+    )
+    properties = model.physical_properties
+    original_properties = {key: value.copy() for key, value in properties.items()}
+    strategy = EngineManagedStrategy(adapter)
+    mocker.patch.object(adapter, "columns", return_value={"A": exp.DataType.build("INT")})
+    if operation == "create_annotated":
+        model = model.copy(update={"columns_to_types_": {"A": exp.DataType.build("INT")}})
+    if operation == "insert":
+        strategy.insert(
+            "test_table",
+            model.render_query_or_raise(),
+            model,
+            True,
+            {},
+            deployability_index=DeployabilityIndex.none_deployable(),
+            snapshot=mocker.Mock(),
+            physical_properties=properties,
+        )
+    else:
+        strategy.create(
+            "test_table",
+            model,
+            operation == "create_managed",
+            {},
+            skip_grants=True,
+            is_snapshot_deployable=operation == "create_managed",
+            physical_properties=properties,
+        )
+    shared_properties = "DATA_RETENTION_TIME_IN_DAYS=1 MAX_DATA_EXTENSION_TIME_IN_DAYS=2"
+    query = 'SELECT "A" AS "A" FROM "SOURCE_TABLE" AS "SOURCE_TABLE"'
+    if operation == "create_annotated":
+        expected_sql = [
+            f'CREATE TABLE IF NOT EXISTS "test_table" ("A" INT) {shared_properties}',
+            f"{query} WHERE FALSE LIMIT 0",
+        ]
+    elif operation == "create_ctas":
+        expected_sql = [
+            f'CREATE TABLE IF NOT EXISTS "test_table" {shared_properties} AS {query} WHERE FALSE LIMIT 0'
+        ]
+    elif operation == "insert":
+        expected_sql = [
+            f'CREATE OR REPLACE TABLE "test_table" {shared_properties} AS SELECT CAST("A" AS INT) AS "A" FROM ({query}) AS "_subquery"'
+        ]
+    else:
+        expected_sql = [
+            'CREATE OR REPLACE DYNAMIC TABLE "test_table" '
+            "WAREHOUSE='SMALL' TARGET_LAG='10 minutes' REFRESH_MODE='AUTO' "
+            "INITIALIZE='ON_CREATE' INITIALIZATION_WAREHOUSE='INITIAL_WH' "
+            f"{shared_properties} AS {query}"
+        ]
+    assert to_sql_calls(adapter) == expected_sql
+    assert properties == original_properties
+
+
 def test_set_current_catalog(make_mocked_engine_adapter: t.Callable):
     adapter = make_mocked_engine_adapter(SnowflakeEngineAdapter)
     adapter._default_catalog = "foo"