Rework DYNAMIC_TABLE, use SHOW AS RESOURCE DYNAMIC TABLE for metadata checks; Implement checks for is_transient, retention_time, column comments, refresh_mode

Author: littleK0i

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## [0.55.0] - 2025-06-26
+
+- Reworked `DYNAMIC_TABLE` to use `SHOW AS RESOURCE DYNAMIC TABLES` command.
+- Implemented additional dynamic table checks for column comments, transient, retention_time, refresh_mode.
+- Added explicit "replace reasons" comments for `DYNAMIC_TABLE`, similar to `TABLE`.
+- `TABLE` comments can now be fully "unset" to NULL instead of empty string.
+
 ## [0.54.2] - 2025-06-25
 
 - Added check for `cluster_by` parameter for existing dynamic tables (thanks to gcv-epalmer).

snowddl/parser/dynamic_table.py

@@ -1,3 +1,5 @@
+from functools import partial
+
 from snowddl.blueprint import (
     AccountObjectIdent,
     DynamicTableBlueprint,
@@ -16,6 +18,7 @@
     RowAccessPolicyReference,
 )
 from snowddl.parser.abc_parser import AbstractParser, ParsedFile
+from snowddl.parser.schema import database_json_schema, schema_json_schema
 
 
 # fmt: off
@@ -155,9 +158,25 @@ class DynamicTableParser(AbstractParser):
     unit_plural_to_singular = {v: k for k, v in unit_singular_to_plural.items()}
 
     def load_blueprints(self):
-        self.parse_schema_object_files("dynamic_table", dynamic_table_json_schema, self.process_dynamic_table)
+        combined_params = {}
+
+        for database_name in self.get_database_names():
+            database_params = self.parse_single_entity_file(f"{database_name}/params", database_json_schema)
+            combined_params[database_name] = {}
+
+            for schema_name in self.get_schema_names_in_database(database_name):
+                schema_params = self.parse_single_entity_file(f"{database_name}/{schema_name}/params", schema_json_schema)
+
+                combined_params[database_name][schema_name] = {
+                    "is_transient": database_params.get("is_transient", False) or schema_params.get("is_transient", False),
+                    "retention_time": schema_params.get("retention_time"),
+                }
+
+        self.parse_schema_object_files(
+            "dynamic_table", dynamic_table_json_schema, partial(self.process_dynamic_table, combined_params=combined_params)
+        )
 
-    def process_dynamic_table(self, f: ParsedFile):
+    def process_dynamic_table(self, f: ParsedFile, combined_params: dict):
         column_blueprints = []
 
         for col_name, col_comment in f.params.get("columns", {}).items():
@@ -177,8 +196,8 @@ def process_dynamic_table(self, f: ParsedFile):
             refresh_mode=f.params.get("refresh_mode").upper() if f.params.get("refresh_mode") else None,
             initialize=f.params.get("initialize").upper() if f.params.get("initialize") else None,
             cluster_by=f.params.get("cluster_by"),
-            is_transient=f.params.get("is_transient", False),
-            retention_time=f.params.get("retention_time"),
+            is_transient=f.params.get("is_transient", combined_params[f.database][f.schema].get("is_transient", False)),
+            retention_time=f.params.get("retention_time", combined_params[f.database][f.schema].get("retention_time", None)),
             depends_on=set(
                 build_schema_object_ident(self.env_prefix, d, f.database, f.schema) for d in f.params.get("depends_on", [])
             ),

snowddl/resolver/dynamic_table.py

@@ -1,44 +1,54 @@
-from re import compile
+from json import loads as json_loads
 
 from snowddl.blueprint import DynamicTableBlueprint
 from snowddl.resolver.abc_schema_object_resolver import AbstractSchemaObjectResolver, ResolveResult, ObjectType
 
-cluster_by_syntax_re = compile(r"^(\w+)?\((.*)\)$")
-
 
 class DynamicTableResolver(AbstractSchemaObjectResolver):
-    # Dynamic tables are available for all accounts during preview, including STANDARD edition
-    # skip_min_edition = Edition.ENTERPRISE
     skip_on_empty_blueprints = True
 
+    unit_to_seconds_multiplier = {
+        "second": 1,
+        "seconds": 1,
+        "minute": 60,
+        "minutes": 60,
+        "hour": 3600,
+        "hours": 3600,
+        "day": 86400,
+        "days": 86400,
+    }
+
     def get_object_type(self) -> ObjectType:
         return ObjectType.DYNAMIC_TABLE
 
     def get_existing_objects_in_schema(self, schema: dict):
         existing_objects = {}
 
         cur = self.engine.execute_meta(
-            "SHOW DYNAMIC TABLES IN SCHEMA {database:i}.{schema:i}",
+            "SHOW AS RESOURCE DYNAMIC TABLES IN SCHEMA {database:i}.{schema:i}",
             {
                 "database": schema["database"],
                 "schema": schema["schema"],
             },
         )
 
         for r in cur:
+            r = json_loads(r["As Resource"])
+
             existing_objects[f"{r['database_name']}.{r['schema_name']}.{r['name']}"] = {
                 "database": r["database_name"],
                 "schema": r["schema_name"],
                 "name": r["name"],
                 "owner": r["owner"],
-                # Extract SQL query text only, skip the initial "CREATE DYNAMIC TABLE ..." part
-                # Snowflake modifies original SQL text in this column, it cannot be compared directly
-                "text": r["text"].partition("\nAS\n")[2].rstrip(";"),
-                "cluster_by": r["cluster_by"] if r["cluster_by"] else None,
+                "is_transient": r["kind"] == "TRANSIENT",
+                "retention_time": r["data_retention_time_in_days"],
+                "columns": r["columns"],
+                "text": r["query"].rstrip(";"),
+                "cluster_by": r["cluster_by"],
                 "target_lag": r["target_lag"],
                 "refresh_mode": r["refresh_mode"],
                 "warehouse": r["warehouse"],
-                "comment": r["comment"] if r["comment"] else None,
+                "comment": r["comment"],
             }
 
         return existing_objects
@@ -66,16 +76,25 @@ def create_object(self, bp: DynamicTableBlueprint):
 
         return ResolveResult.CREATE
 
-    def _compare_cluster_by(self, bp: DynamicTableBlueprint, row: dict):
-        bp_cluster_by = ", ".join(bp.cluster_by) if bp.cluster_by else None
-        snow_cluster_by = cluster_by_syntax_re.sub(r"\2", row["cluster_by"]) if row["cluster_by"] else None
-
-        return bp_cluster_by == snow_cluster_by
-
     def compare_object(self, bp: DynamicTableBlueprint, row: dict):
         result = ResolveResult.NOCHANGE
+        replace_reasons = []
+
+        if bp.columns and [str(c.name) for c in bp.columns] != [str(c["name"]) for c in row["columns"]]:
+            replace_reasons.append("Column definition was changed")
 
         if bp.text != row["text"]:
+            replace_reasons.append("SQL text was changed")
+
+        if bp.is_transient is True and row["is_transient"] is False:
+            replace_reasons.append("Dynamic table type was changed to TRANSIENT")
+        elif bp.is_transient is False and row["is_transient"] is True:
+            replace_reasons.append("Dynamic table type was changed to PERMANENT")
+
+        if bp.refresh_mode and bp.refresh_mode != "AUTO" and bp.refresh_mode != row["refresh_mode"]:
+            replace_reasons.append(f"Refresh mode was changed to {bp.refresh_mode}")
+
+        if replace_reasons:
             query = self.engine.query_builder()
             query.append("CREATE OR REPLACE")
 
@@ -90,11 +109,11 @@ def compare_object(self, bp: DynamicTableBlueprint, row: dict):
             )
 
             query.append(self._build_common_dynamic_table_sql(bp))
-            self.engine.execute_unsafe_ddl(query)
+            self.engine.execute_unsafe_ddl("\n".join(f"-- {r}" for r in replace_reasons) + "\n" + str(query))
 
             return ResolveResult.REPLACE
 
-        if bp.target_lag != row["target_lag"]:
+        if not self._compare_target_lag(bp, row):
             self.engine.execute_safe_ddl(
                 "ALTER DYNAMIC TABLE {full_name:i} SET TARGET_LAG = {target_lag}",
                 {
@@ -133,6 +152,17 @@ def compare_object(self, bp: DynamicTableBlueprint, row: dict):
 
             result = ResolveResult.ALTER
 
+        if bp.retention_time is not None and bp.retention_time != row["retention_time"]:
+            self.engine.execute_unsafe_ddl(
+                "ALTER DYNAMIC TABLE {full_name:i} SET DATA_RETENTION_TIME_IN_DAYS = {retention_time:d}",
+                {
+                    "full_name": bp.full_name,
+                    "retention_time": bp.retention_time,
+                },
+            )
+
+            result = ResolveResult.ALTER
+
         if bp.comment != row["comment"]:
             self.engine.execute_safe_ddl(
                 "ALTER DYNAMIC TABLE {full_name:i} SET COMMENT = {comment}",
@@ -144,6 +174,30 @@ def compare_object(self, bp: DynamicTableBlueprint, row: dict):
 
             result = ResolveResult.ALTER
 
+        for idx, c in enumerate(row["columns"]):
+            bp_col_comment = bp.columns[idx].comment if bp.columns else None
+
+            if bp_col_comment != c["comment"]:
+                if bp_col_comment:
+                    self.engine.execute_safe_ddl(
+                        "ALTER DYNAMIC TABLE {full_name:i} MODIFY COLUMN {column_name:i} COMMENT {comment}",
+                        {
+                            "full_name": bp.full_name,
+                            "column_name": c["name"],
+                            "comment": bp_col_comment,
+                        },
+                    )
+                else:
+                    self.engine.execute_safe_ddl(
+                        "ALTER DYNAMIC TABLE {full_name:i} MODIFY COLUMN {column_name:i} UNSET COMMENT",
+                        {
+                            "full_name": bp.full_name,
+                            "column_name": c["name"],
+                        },
+                    )
+
+                result = ResolveResult.ALTER
+
         return result
 
     def drop_object(self, row: dict):
@@ -241,3 +295,25 @@ def _build_common_dynamic_table_sql(self, bp: DynamicTableBlueprint):
         query.append_nl(bp.text)
 
         return query
+
+    def _compare_cluster_by(self, bp: DynamicTableBlueprint, row: dict):
+        bp_cluster_by = ", ".join(bp.cluster_by).upper() if bp.cluster_by else None
+        snow_cluster_by = ", ".join(row["cluster_by"]).upper() if row["cluster_by"] else None
+
+        return bp_cluster_by == snow_cluster_by
+
+    def _compare_target_lag(self, bp: DynamicTableBlueprint, row: dict):
+        if bp.target_lag == "DOWNSTREAM":
+            return row["target_lag"]["type"] == "DOWNSTREAM"
+
+        num, _, unit = bp.target_lag.partition(" ")
+
+        num = int(num)
+        unit = unit.lower()
+
+        num_in_seconds = num * self.unit_to_seconds_multiplier[unit]
+
+        if row["target_lag"]["type"] == "USER_DEFINED" and row["target_lag"]["seconds"] == num_in_seconds:
+            return True
+
+        return False

snowddl/resolver/table.py

@@ -165,16 +165,25 @@ def compare_object(self, bp: TableBlueprint, row: dict):
 
             # Comments
             if snow_c.comment != bp_c.comment:
-                # UNSET COMMENT is currently not supported for columns, we can only set it to empty string
-                safe_alters.append(
-                    self.engine.format(
-                        "MODIFY COLUMN {col_name:i} COMMENT {comment}",
-                        {
-                            "col_name": col_name,
-                            "comment": bp_c.comment if bp_c.comment else "",
-                        },
+                if bp_c.comment:
+                    safe_alters.append(
+                        self.engine.format(
+                            "MODIFY COLUMN {col_name:i} COMMENT {comment}",
+                            {
+                                "col_name": col_name,
+                                "comment": bp_c.comment,
+                            },
+                        )
+                    )
+                else:
+                    safe_alters.append(
+                        self.engine.format(
+                            "MODIFY COLUMN {col_name:i} UNSET COMMENT",
+                            {
+                                "col_name": col_name,
+                            },
+                        )
                     )
-                )
 
             # If type matches exactly, skip all other checks
             if snow_c.type == bp_c.type:
@@ -266,7 +275,7 @@ def compare_object(self, bp: TableBlueprint, row: dict):
         if bp.is_transient is True and row["is_transient"] is False:
             replace_reasons.append("Table type was changed to TRANSIENT")
         elif bp.is_transient is False and row["is_transient"] is True:
-            replace_reasons.append("Table type was changed to no longer being TRANSIENT")
+            replace_reasons.append("Table type was changed to PERMANENT")
 
         # Retention time
         if bp.retention_time is not None and bp.retention_time != row["retention_time"]:

snowddl/version.py

@@ -1 +1 @@
-__version__ = "0.54.2"
+__version__ = "0.55.0"

test/_config/step1/db1/sc1/dynamic_table/dt004_dt1.yaml

@@ -1,3 +1,7 @@
+columns:
+  id: aaa
+  name: ~
+
 text: |-
   SELECT id, name
   FROM dt004_tb1

test/_config/step2/db1/sc1/dynamic_table/dt004_dt1.yaml

@@ -1,3 +1,7 @@
+columns:
+  id: ~
+  name: bbb
+
 text: |-
   SELECT id, name
   FROM dt004_tb1

test/dynamic_table/dt004.py

@@ -1,19 +1,31 @@
 def test_step1(helper):
     dynamic_table_show = helper.show_dynamic_table("db1", "sc1", "dt004_dt1")
+    dynamic_table_desc = helper.desc_table("db1", "sc1", "dt004_dt1")
 
     assert dynamic_table_show["automatic_clustering"] == "ON"
     assert dynamic_table_show["cluster_by"] == "LINEAR(id)"
 
+    assert dynamic_table_desc["ID"]["comment"] == "aaa"
+    assert dynamic_table_desc["NAME"]["comment"] is None
+
 
 def test_step2(helper):
     dynamic_table_show = helper.show_dynamic_table("db1", "sc1", "dt004_dt1")
+    dynamic_table_desc = helper.desc_table("db1", "sc1", "dt004_dt1")
 
     assert dynamic_table_show["automatic_clustering"] == "ON"
     assert dynamic_table_show["cluster_by"] == "LINEAR(id, name)"
 
+    assert dynamic_table_desc["ID"]["comment"] is None
+    assert dynamic_table_desc["NAME"]["comment"] == "bbb"
+
 
 def test_step3(helper):
     dynamic_table_show = helper.show_dynamic_table("db1", "sc1", "dt004_dt1")
+    dynamic_table_desc = helper.desc_table("db1", "sc1", "dt004_dt1")
 
     assert dynamic_table_show["automatic_clustering"] == "OFF"
     assert dynamic_table_show["cluster_by"] == ""
+
+    assert dynamic_table_desc["ID"]["comment"] is None
+    assert dynamic_table_desc["NAME"]["comment"] is None