Rework and extend drop intention cache to include columns; Prevent generation of DROP PRIMARY KEY, DROP UNIQUE KEY, DROP FOREIGN KEY commands if relevant tables or columns are about to be dropped

Author: littleK0i

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## [0.56.0] - 2025-09-02
+
+- Reworked and extended drop intention cache logic to table columns.
+- Prevented generation of `DROP PRIMARY KEY`, `DROP UNIQUE KEY` and `DROP FOREIGN KEY` commands if relevant tables or columns are about to be dropped.
+- Removed unused resolver for `TAG` object type.
+
 ## [0.55.1] - 2025-07-23
 
 - Prevented errors for outbound shares with organization listings.

snowddl/app/base.py

@@ -722,7 +722,7 @@ def output_engine_stats(self, engine: SnowDDLEngine):
         self.logger.info(f"Executed {len(engine.executed_ddl)} DDL queries, Suggested {len(engine.suggested_ddl)} DDL queries")
 
     def output_engine_warnings(self, engine: SnowDDLEngine):
-        for object_type, object_names in engine.intention_cache.invalid_name_warning.items():
+        for object_type, object_names in engine.intention_cache.object_name_warning.items():
             for name in object_names:
                 self.logger.warning(
                     f"Detected {object_type.name} with name [{name}] "

snowddl/cache/intention_cache.py

@@ -20,24 +20,24 @@ class IntentionCache:
     def __init__(self, engine: "SnowDDLEngine"):
         self.engine = engine
 
-        self.drop_intention: Dict[ObjectType, Set[str]] = defaultdict(set)
-        self.replace_intention: Dict[ObjectType, Set[str]] = defaultdict(set)
+        self.object_drop_intention: Dict[ObjectType, Set[str]] = defaultdict(set)
+        self.object_name_warning: Dict[ObjectType, Set[str]] = defaultdict(set)
 
-        self.invalid_name_warning: Dict[ObjectType, Set[str]] = defaultdict(set)
+        self.column_drop_intention: Dict[str, Set[str]] = defaultdict(set)
 
-    def add_drop_intention(self, object_type: ObjectType, object_full_name: str):
-        self.drop_intention[object_type].add(object_full_name)
+    def add_object_drop_intention(self, object_type: ObjectType, object_full_name: str):
+        self.object_drop_intention[object_type].add(object_full_name)
 
-    def add_replace_intention(self, object_type: ObjectType, object_full_name: str):
-        self.replace_intention[object_type].add(object_full_name)
+    def add_object_name_warning(self, object_type: ObjectType, object_full_name: str):
+        self.object_name_warning[object_type].add(object_full_name)
 
-    def add_invalid_name_warning(self, object_type: ObjectType, object_full_name: str):
-        self.invalid_name_warning[object_type].add(object_full_name)
+    def add_column_drop_intention(self, object_full_name: str, column_name: str):
+        self.column_drop_intention[object_full_name].add(column_name)
 
-    def check_drop_intention(self, object_type: ObjectType, object_full_name: str):
-        return object_full_name in self.drop_intention[object_type]
+    def check_object_drop_intention(self, object_type: ObjectType, object_full_name: str):
+        return object_full_name in self.object_drop_intention[object_type]
 
-    def check_parent_drop_intention(self, object_type: ObjectType, object_full_name: str):
+    def check_parent_object_drop_intention(self, object_type: ObjectType, object_full_name: str):
         blueprint_cls = object_type.blueprint_cls
         object_full_name_parts = object_full_name.partition("(")[0].split(".")
 
@@ -47,24 +47,27 @@ def check_parent_drop_intention(self, object_type: ObjectType, object_full_name:
 
         # All schemas and schema objects are implicitly dropped by DATABASE
         if issubclass(blueprint_cls, (SchemaBlueprint, SchemaObjectBlueprint, DatabaseRoleBlueprint)) and (
-            database_name in self.drop_intention[ObjectType.DATABASE]
+            database_name in self.object_drop_intention[ObjectType.DATABASE]
         ):
             return True
 
         # All schema objects are implicitly dropped by SCHEMA
-        if issubclass(blueprint_cls, SchemaObjectBlueprint) and (schema_name in self.drop_intention[ObjectType.SCHEMA]):
+        if issubclass(blueprint_cls, SchemaObjectBlueprint) and (schema_name in self.object_drop_intention[ObjectType.SCHEMA]):
             return True
 
         # All stage files are implicitly dropped by STAGE
-        if issubclass(blueprint_cls, StageFileBlueprint) and (schema_object_name in self.drop_intention[ObjectType.STAGE]):
+        if issubclass(blueprint_cls, StageFileBlueprint) and (schema_object_name in self.object_drop_intention[ObjectType.STAGE]):
             return True
 
         # All table constraints are implicitly dropped by various TABLE types which support constraints
         if issubclass(blueprint_cls, (ForeignKeyBlueprint, PrimaryKeyBlueprint, UniqueKeyBlueprint)) and (
-            schema_object_name in self.drop_intention[ObjectType.TABLE]
-            or schema_object_name in self.drop_intention[ObjectType.EXTERNAL_TABLE]
-            or schema_object_name in self.drop_intention[ObjectType.HYBRID_TABLE]
+            schema_object_name in self.object_drop_intention[ObjectType.TABLE]
+            or schema_object_name in self.object_drop_intention[ObjectType.EXTERNAL_TABLE]
+            or schema_object_name in self.object_drop_intention[ObjectType.HYBRID_TABLE]
         ):
             return True
 
         return False
+
+    def check_column_drop_intention(self, object_full_name: str, column_name: str):
+        return column_name in self.column_drop_intention[object_full_name]

snowddl/resolver/abc_resolver.py

@@ -95,8 +95,8 @@ def _resolve_drop(self):
             if object_full_name in self.blueprints:
                 continue
 
-            # Parent object is going to be dropped
-            if self.engine.intention_cache.check_parent_drop_intention(self.object_type, object_full_name):
+            # Another object is going to be dropped, which implicitly drops this object
+            if self._check_implicit_drop_intention(object_full_name):
                 continue
 
             tasks[object_full_name] = (self.drop_object, self.existing_objects[object_full_name])
@@ -107,15 +107,22 @@ def _destroy_drop(self):
         tasks = {}
 
         # Drop all existing objects
-        for full_name in sorted(self.existing_objects):
-            # Parent object is going to be dropped
-            if self.engine.intention_cache.check_parent_drop_intention(self.object_type, full_name):
+        for object_full_name in sorted(self.existing_objects):
+            # Another object is going to be dropped, which implicitly drops this object
+            if self._check_implicit_drop_intention(object_full_name):
                 continue
 
-            tasks[full_name] = (self.drop_object, self.existing_objects[full_name])
+            tasks[object_full_name] = (self.drop_object, self.existing_objects[object_full_name])
 
         self._process_tasks(tasks)
 
+    def _check_implicit_drop_intention(self, object_full_name: str) -> bool:
+        # Dropping or replacing any parent object implicitly drops this object
+        if self.engine.intention_cache.check_parent_object_drop_intention(self.object_type, object_full_name):
+            return True
+
+        return False
+
     def _process_tasks(self, tasks):
         futures = {}
 
@@ -128,11 +135,8 @@ def _process_tasks(self, tasks):
             try:
                 result = f.result()
 
-                if result == ResolveResult.REPLACE:
-                    self.engine.intention_cache.add_replace_intention(self.object_type, full_name)
-
-                if result == ResolveResult.DROP:
-                    self.engine.intention_cache.add_drop_intention(self.object_type, full_name)
+                if result in (ResolveResult.REPLACE, ResolveResult.DROP):
+                    self.engine.intention_cache.add_object_drop_intention(self.object_type, full_name)
 
                 if result == ResolveResult.NOCHANGE:
                     self.engine.logger.debug(f"Resolved {self.object_type.name} [{full_name}]: {result.value}")

snowddl/resolver/abc_role_resolver.py

@@ -99,7 +99,7 @@ def get_existing_role_grants(self, role_name):
                 try:
                     grant_name = build_grant_name_ident(self.config.env_prefix, r["name"], object_type)
                 except (KeyError, ValueError):
-                    self.engine.intention_cache.add_invalid_name_warning(object_type, r["name"])
+                    self.engine.intention_cache.add_object_name_warning(object_type, r["name"])
                     continue
 
                 grants.append(
@@ -128,7 +128,7 @@ def get_existing_role_grants(self, role_name):
             try:
                 grant_name = build_future_grant_name_ident(object_type, r["name"])
             except ValueError:
-                self.engine.intention_cache.add_invalid_name_warning(object_type, r["name"])
+                self.engine.intention_cache.add_object_name_warning(object_type, r["name"])
                 continue
 
             future_grants.append(

snowddl/resolver/abc_schema_object_resolver.py

@@ -28,8 +28,8 @@ def _resolve_drop(self):
             if object_full_name in self.blueprints:
                 continue
 
-            # Parent object is going to be dropped
-            if self.engine.intention_cache.check_parent_drop_intention(self.object_type, object_full_name):
+            # Another object is going to be dropped, which implicitly drops this object
+            if self._check_implicit_drop_intention(object_full_name):
                 continue
 
             schema_full_name = ".".join(object_full_name.split(".")[:2])

snowddl/resolver/foreign_key.py

@@ -107,3 +107,33 @@ def drop_object(self, row: dict):
         )
 
         return ResolveResult.DROP
+
+    def _check_implicit_drop_intention(self, object_full_name: str) -> bool:
+        # Dropping any of parent objects implicitly drops this object
+        if self.engine.intention_cache.check_parent_object_drop_intention(self.object_type, object_full_name):
+            return True
+
+        fk = self.existing_objects[object_full_name]
+
+        for ref_table_object_type in [ObjectType.TABLE, ObjectType.EXTERNAL_TABLE]:
+            # Dropping ref_table implicitly drops all foreign keys pointing to it
+            if self.engine.intention_cache.check_object_drop_intention(ref_table_object_type, fk["ref_table_name"]):
+                return True
+
+            # Dropping any of parent objects of ref_table implicitly drops this object
+            if self.engine.intention_cache.check_parent_object_drop_intention(ref_table_object_type, fk["ref_table_name"]):
+                return True
+
+        table_full_name = f"{fk['database']}.{fk['schema']}.{fk['table']}"
+
+        # Dropping one of foreign key columns implicitly drops the entire key
+        for col_name in fk["columns"]:
+            if self.engine.intention_cache.check_column_drop_intention(table_full_name, col_name):
+                return True
+
+        # Dropping one of foreign key columns in ref_table implicitly drops the entire key
+        for col_name in fk["ref_columns"]:
+            if self.engine.intention_cache.check_column_drop_intention(fk["ref_table_name"], col_name):
+                return True
+
+        return False

snowddl/resolver/primary_key.py

@@ -92,3 +92,15 @@ def drop_object(self, row: dict):
         )
 
         return ResolveResult.DROP
+
+    def _check_implicit_drop_intention(self, object_full_name: str) -> bool:
+        # Dropping any of parent objects implicitly drops the entire key
+        if self.engine.intention_cache.check_parent_object_drop_intention(self.object_type, object_full_name):
+            return True
+
+        # Dropping one of primary key columns implicitly drops the entire key
+        for col_name in self.existing_objects[object_full_name]["columns"]:
+            if self.engine.intention_cache.check_column_drop_intention(object_full_name, col_name):
+                return True
+
+        return False

snowddl/resolver/schema.py

@@ -108,7 +108,7 @@ def _resolve_drop(self):
                 continue
 
             # Parent object is going to be dropped
-            if self.engine.intention_cache.check_parent_drop_intention(self.object_type, schema_full_name):
+            if self.engine.intention_cache.check_parent_object_drop_intention(self.object_type, schema_full_name):
                 continue
 
             database_full_name = ".".join(schema_full_name.split(".")[:1])

snowddl/resolver/table.py

@@ -79,6 +79,7 @@ def compare_object(self, bp: TableBlueprint, row: dict):
         bp_cols = {str(c.name): c for c in bp.columns}
         snow_cols = self._get_existing_columns(bp)
 
+        dropping_col_names = []
         remaining_col_names = list(snow_cols.keys())
 
         for col_name, snow_c in snow_cols.items():
@@ -93,7 +94,9 @@ def compare_object(self, bp: TableBlueprint, row: dict):
                     )
                 )
 
+                dropping_col_names.append(col_name)
                 remaining_col_names.remove(col_name)
+
                 replace_notices.append(f"Column {col_name} is about to be dropped")
                 continue
 
@@ -354,6 +357,9 @@ def compare_object(self, bp: TableBlueprint, row: dict):
 
             result = ResolveResult.ALTER
 
+            for col_name in dropping_col_names:
+                self.engine.intention_cache.add_column_drop_intention(str(bp.full_name), col_name)
+
         # If table was re-created, apply or suggest search optimization using exactly the same condition value
         if result == ResolveResult.REPLACE:
             self._create_search_optimization(bp, condition=self.engine.settings.execute_replace_table)