Implement SNAPSHOT_POLICY, SNAPSHOT_SET; Add workaround for user TYPE

Author: littleK0i

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## [0.57.0] - 2025-09-16
+
+- Introduced workaround for user type `NULL` being replaced with `PERSON` in bundle `2025_05`.
+- Implemented initial logic for `SNAPSHOT POLICY` and `SNAPSHOT SET` object types.
+
 ## [0.56.0] - 2025-09-02
 
 - Reworked and extended drop intention cache logic to table columns.

snowddl/blueprint/__init__.py

@@ -35,6 +35,8 @@
     SecretBlueprint,
     SemanticViewBlueprint,
     SequenceBlueprint,
+    SnapshotPolicyBlueprint,
+    SnapshotSetBlueprint,
     StageBlueprint,
     StageFileBlueprint,
     StreamBlueprint,

snowddl/blueprint/blueprint.py

@@ -374,6 +374,17 @@ class SequenceBlueprint(SchemaObjectBlueprint):
     is_ordered: Optional[bool] = None
 
 
+class SnapshotPolicyBlueprint(SchemaObjectBlueprint):
+    schedule: Optional[str] = None
+    expire_after_days: Optional[int] = None
+
+
+class SnapshotSetBlueprint(SchemaObjectBlueprint):
+    object_type: ObjectType
+    object_name: Union[DatabaseIdent, SchemaIdent, SchemaObjectIdent]
+    snapshot_policy: Optional[SchemaObjectIdent] = None
+
+
 class StageBlueprint(SchemaObjectBlueprint):
     url: Optional[str] = None
     storage_integration: Optional[Ident] = None

snowddl/blueprint/object_type.py

@@ -247,6 +247,18 @@ class ObjectType(Enum):
         "blueprint_cls": "OutboundShareBlueprint",
     }
 
+    SNAPSHOT_POLICY = {
+        "singular": "SNAPSHOT POLICY",
+        "plural": "SNAPSHOT POLICIES",
+        "blueprint_cls": "SnapshotPolicyBlueprint",
+    }
+
+    SNAPSHOT_SET = {
+        "singular": "SNAPSHOT SET",
+        "plural": "SNAPSHOT SETS",
+        "blueprint_cls": "SnapshotSetBlueprint",
+    }
+
     STAGE = {
         "singular": "STAGE",
         "plural": "STAGES",

snowddl/parser/__init__.py

@@ -34,6 +34,8 @@
 from .secret import SecretParser
 from .semantic_view import SemanticViewParser
 from .sequence import SequenceParser
+from .snapshot_policy import SnapshotPolicyParser
+from .snapshot_set import SnapshotSetParser
 from .stage import StageParser
 from .stream import StreamParser
 from .table import TableParser
@@ -55,6 +57,8 @@
     RowAccessPolicyParser,
     ResourceMonitorParser,
     AccountPolicyParser,
+    SnapshotPolicyParser,
+    SnapshotSetParser,
     # --
     WarehouseParser,
     DatabaseParser,
@@ -95,6 +99,8 @@
     MaskingPolicyParser,
     ProjectionPolicyParser,
     RowAccessPolicyParser,
+    SnapshotPolicyParser,
+    SnapshotSetParser,
     # --
     DatabaseParser,
     SchemaParser,

snowddl/parser/snapshot_policy.py

@@ -0,0 +1,36 @@
+from snowddl.blueprint import SnapshotPolicyBlueprint, SchemaObjectIdent
+from snowddl.parser.abc_parser import AbstractParser, ParsedFile
+
+
+# fmt: off
+snapshot_policy_json_schema = {
+    "type": "object",
+    "properties": {
+        "schedule": {
+            "type": "string"
+        },
+        "expire_after_days": {
+            "type": "integer"
+        },
+        "comment": {
+            "type": "string"
+        }
+    },
+    "additionalProperties": False
+}
+# fmt: on
+
+
+class SnapshotPolicyParser(AbstractParser):
+    def load_blueprints(self):
+        self.parse_schema_object_files("snapshot_policy", snapshot_policy_json_schema, self.process_snapshot_policy)
+
+    def process_snapshot_policy(self, f: ParsedFile):
+        bp = SnapshotPolicyBlueprint(
+            full_name=SchemaObjectIdent(self.env_prefix, f.database, f.schema, f.name),
+            schedule=f.params.get("schedule"),
+            expire_after_days=f.params.get("expire_after_days"),
+            comment=f.params.get("comment"),
+        )
+
+        self.config.add_blueprint(bp)

snowddl/parser/snapshot_set.py

@@ -0,0 +1,56 @@
+from snowddl.blueprint import (
+    DatabaseIdent,
+    SchemaIdent,
+    ObjectType,
+    SnapshotSetBlueprint,
+    SchemaObjectIdent,
+    build_schema_object_ident,
+)
+from snowddl.parser.abc_parser import AbstractParser, ParsedFile
+
+
+# fmt: off
+snapshot_set_json_schema = {
+    "type": "object",
+    "properties": {
+        "object_type": {
+            "type": "string"
+        },
+        "object_name": {
+            "type": "string"
+        },
+        "snapshot_policy": {
+            "type": "string"
+        },
+        "comment": {
+            "type": "string"
+        }
+    },
+    "additionalProperties": False
+}
+# fmt: on
+
+
+class SnapshotSetParser(AbstractParser):
+    def load_blueprints(self):
+        self.parse_schema_object_files("snapshot_set", snapshot_set_json_schema, self.process_snapshot_set)
+
+    def process_snapshot_set(self, f: ParsedFile):
+        object_type = ObjectType[f.params["object_type"]]
+
+        if object_type == ObjectType.DATABASE:
+            object_name = DatabaseIdent(self.env_prefix, f.params["object_name"])
+        elif object_type == ObjectType.SCHEMA:
+            object_name = SchemaIdent(self.env_prefix, *f.params["object_name"].split(".", 2))
+        else:
+            object_name = build_schema_object_ident(self.env_prefix, f.params["object_name"], f.database, f.schema)
+
+        bp = SnapshotSetBlueprint(
+            full_name=SchemaObjectIdent(self.env_prefix, f.database, f.schema, f.name),
+            object_type=ObjectType[f.params["object_type"]],
+            object_name=object_name,
+            snapshot_policy=build_schema_object_ident(self.env_prefix, f.params["snapshot_policy"], f.database, f.schema) if f.params.get("snapshot_policy") else None,
+            comment=f.params.get("comment"),
+        )
+
+        self.config.add_blueprint(bp)

snowddl/resolver/__init__.py

@@ -41,6 +41,8 @@
 from .schema_write_role import SchemaWriteRoleResolver
 from .secret import SecretResolver
 from .semantic_view import SemanticViewResolver
+from .snapshot_policy import SnapshotPolicyResolver
+from .snapshot_set import SnapshotSetResolver
 from .stage import StageResolver
 from .stage_file import StageFileResolver
 from .stream import StreamResolver
@@ -112,6 +114,8 @@
     NetworkPolicyResolver,
     ProjectionPolicyResolver,
     RowAccessPolicyResolver,
+    SnapshotPolicyResolver,
+    SnapshotSetResolver,
 ]
 
 
@@ -179,6 +183,8 @@
     MaskingPolicyResolver,
     ProjectionPolicyResolver,
     RowAccessPolicyResolver,
+    SnapshotPolicyResolver,
+    SnapshotSetResolver,
 ]
 
 

snowddl/resolver/snapshot_policy.py

@@ -0,0 +1,140 @@
+from snowddl.blueprint import SnapshotPolicyBlueprint
+from snowddl.resolver.abc_schema_object_resolver import AbstractSchemaObjectResolver, ResolveResult, ObjectType
+
+
+class SnapshotPolicyResolver(AbstractSchemaObjectResolver):
+    skip_on_empty_blueprints = True
+
+    def get_object_type(self) -> ObjectType:
+        return ObjectType.SNAPSHOT_POLICY
+
+    def get_existing_objects_in_schema(self, schema: dict):
+        existing_objects = {}
+
+        cur = self.engine.execute_meta(
+            "SHOW SNAPSHOT POLICIES IN SCHEMA {database:i}.{schema:i}",
+            {
+                "database": schema["database"],
+                "schema": schema["schema"],
+            },
+        )
+
+        for r in cur:
+            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"],
+                "schedule": r["schedule"],
+                "expire_after_days": r["expire_after_days"],
+                "comment": r["comment"] if r["comment"] else None,
+            }
+
+        return existing_objects
+
+    def get_blueprints(self):
+        return self.config.get_blueprints_by_type(SnapshotPolicyBlueprint)
+
+    def create_object(self, bp: SnapshotPolicyBlueprint):
+        query = self.engine.query_builder()
+
+        query.append(
+            "CREATE SNAPSHOT POLICY {full_name:i}",
+            {
+                "full_name": bp.full_name,
+            },
+        )
+
+        if bp.schedule:
+            query.append_nl(
+                "SCHEDULE = {schedule}",
+                {
+                    "schedule": bp.schedule,
+                },
+            )
+
+        if bp.expire_after_days:
+            query.append_nl(
+                "EXPIRE_AFTER_DAYS = {expire_after_days:d}",
+                {
+                    "expire_after_days": bp.expire_after_days,
+                },
+            )
+
+        if bp.comment:
+            query.append_nl(
+                "COMMENT = {comment}",
+                {
+                    "comment": bp.comment,
+                },
+            )
+
+        self.engine.execute_unsafe_ddl(query)
+
+        return ResolveResult.CREATE
+
+    def compare_object(self, bp: SnapshotPolicyBlueprint, row: dict):
+        result = ResolveResult.NOCHANGE
+
+        if bp.schedule != row["schedule"]:
+            if bp.schedule:
+                self.engine.execute_unsafe_ddl(
+                    "ALTER SNAPSHOT POLICY {full_name:i} SET SCHEDULE = {schedule}",
+                    {
+                        "full_name": bp.full_name,
+                        "schedule": bp.schedule,
+                    },
+                )
+            else:
+                self.engine.execute_unsafe_ddl(
+                    "ALTER SNAPSHOT POLICY {full_name:i} UNSET SCHEDULE",
+                    {
+                        "full_name": bp.full_name,
+                    },
+                )
+
+            result = ResolveResult.ALTER
+
+        if bp.expire_after_days != row["expire_after_days"]:
+            if bp.expire_after_days:
+                self.engine.execute_unsafe_ddl(
+                    "ALTER SNAPSHOT POLICY {full_name:i} SET EXPIRE_AFTER_DAYS = {expire_after_days:d}",
+                    {
+                        "full_name": bp.full_name,
+                        "expire_after_days": bp.expire_after_days,
+                    },
+                )
+            else:
+                self.engine.execute_unsafe_ddl(
+                    "ALTER SNAPSHOT POLICY {full_name:i} UNSET EXPIRE_AFTER_DAYS",
+                    {
+                        "full_name": bp.full_name,
+                    },
+                )
+
+            result = ResolveResult.ALTER
+
+        if bp.comment != row["comment"]:
+            self.engine.execute_unsafe_ddl(
+                "ALTER SNAPSHOT POLICY {full_name:i} SET COMMENT = {comment}",
+                {
+                    "full_name": bp.full_name,
+                    "comment": bp.comment,
+                },
+            )
+
+            result = ResolveResult.ALTER
+
+        return result
+
+    def drop_object(self, row: dict):
+        self.engine.execute_unsafe_ddl(
+            "DROP SNAPSHOT POLICY {database:i}.{schema:i}.{name:i}",
+            {
+                "database": row["database"],
+                "schema": row["schema"],
+                "name": row["name"],
+            },
+        )
+
+        return ResolveResult.DROP