Add CLI option --clone-source-env-prefix

Author: littleK0i

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## [0.53.0] -2025-05-26
+
+- Added policies for `DYNAMIC_TABLE` config.
+- Fixed incorrect object type names while dropping policy references.
+- Added CLI option `--clone-source-env-prefix`. It allowed to clone tables from another environment with env prefix. Previously it was possible to clone only from databases without prefix.
+
 ## [0.52.0] - 2025-05-23
 
 - Reworked data type parsing logic for `SHOW FUNCTIONS` and `SHOW PROCEDURES` to prepare for bundle `2025_03` [changes](https://docs.snowflake.com/en/release-notes/bcr-bundles/2025_03/bcr-1944).

snowddl/app/base.py

@@ -258,10 +258,14 @@ def init_arguments_parser(self):
         # Cloning
         parser.add_argument(
             "--clone-table",
-            help="Clone all tables from source databases (without env_prefix) to destination databases (with env_prefix)",
+            help="Clone all tables from source databases to destination databases (with env_prefix)",
             default=False,
             action="store_true",
         )
+        parser.add_argument(
+            "--clone-source-env-prefix",
+            help="Clone from another environment with different env_prefix",
+        )
 
         # Destroy without env prefix
         parser.add_argument(
@@ -497,6 +501,12 @@ def init_settings(self):
 
             settings.clone_table = True
 
+        if self.args.get("clone_source_env_prefix"):
+            env_prefix = self.args.get("clone_source_env_prefix")
+            env_prefix_separator = self.args.get("env_prefix_separator")
+
+            settings.clone_source_env_prefix = f"{env_prefix}{env_prefix_separator}".upper()
+
         if self.args.get("env_admin_role"):
             settings.env_admin_role = Ident(self.args.get("env_admin_role"))
 

snowddl/app/singledb.py

@@ -216,6 +216,10 @@ def init_arguments_parser(self):
             default=False,
             action="store_true",
         )
+        parser.add_argument(
+            "--clone-source-env-prefix",
+            help="Clone from another environment with different env_prefix",
+        )
 
         # Subparsers
         subparsers = parser.add_subparsers(dest="action")

snowddl/resolver/clone_table.py

@@ -1,4 +1,4 @@
-from snowddl.blueprint import TableBlueprint
+from snowddl.blueprint import TableBlueprint, SchemaObjectIdent
 from snowddl.resolver.abc_schema_object_resolver import AbstractResolver, ResolveResult, ObjectType
 
 
@@ -26,6 +26,7 @@ def get_existing_objects(self):
 
     def get_databases_for_clone(self):
         databases_for_clone = {}
+        clone_source_env_prefix = self.engine.settings.clone_source_env_prefix
 
         cur = self.engine.execute_meta("SHOW DATABASES")
 
@@ -34,12 +35,24 @@ def get_databases_for_clone(self):
             if r["origin"]:
                 continue
 
-            # Skip databases without destination for cloning
-            if f"{self.config.env_prefix}{r['name']}" not in self.engine.schema_cache.databases:
+            src_database = str(r["name"])
+
+            if clone_source_env_prefix:
+                # Skip everything which does not start with source prefix
+                if not src_database.startswith(clone_source_env_prefix):
+                    continue
+
+                dst_database = f"{self.config.env_prefix}{src_database.removeprefix(clone_source_env_prefix)}"
+            else:
+                dst_database = f"{self.config.env_prefix}{src_database}"
+
+            # Skip every source database without destination database for cloning
+            if dst_database not in self.engine.schema_cache.databases:
                 continue
 
-            databases_for_clone[r["name"]] = {
-                "database": r["name"],
+            databases_for_clone[dst_database] = {
+                "src_database": src_database,
+                "dst_database": dst_database,
             }
 
         return databases_for_clone
@@ -48,9 +61,9 @@ def get_schemas_for_clone(self, database):
         schemas_for_clone = {}
 
         cur = self.engine.execute_meta(
-            "SHOW SCHEMAS IN DATABASE {database:i}",
+            "SHOW SCHEMAS IN DATABASE {src_database:i}",
             {
-                "database": database["database"],
+                "src_database": database["src_database"],
             },
         )
 
@@ -59,12 +72,15 @@ def get_schemas_for_clone(self, database):
             if r["name"] == "INFORMATION_SCHEMA":
                 continue
 
-            # Skip schemas without destination for cloning
-            if f"{self.config.env_prefix}{r['database_name']}.{r['name']}" not in self.engine.schema_cache.schemas:
+            dst_schema = f"{database['dst_database']}.{r['name']}"
+
+            # Skip every source schema without destination schema for cloning
+            if dst_schema not in self.engine.schema_cache.schemas:
                 continue
 
-            schemas_for_clone[f"{r['database_name']}.{r['name']}"] = {
-                "database": r["database_name"],
+            schemas_for_clone[dst_schema] = {
+                "src_database": database["src_database"],
+                "dst_database": database["dst_database"],
                 "schema": r["name"],
             }
 
@@ -74,9 +90,9 @@ def get_tables_for_clone(self, schema):
         tables_for_clone = {}
 
         cur = self.engine.execute_meta(
-            "SHOW TABLES IN SCHEMA {database:i}.{schema:i}",
+            "SHOW TABLES IN SCHEMA {src_database:i}.{schema:i}",
             {
-                "database": schema["database"],
+                "src_database": schema["src_database"],
                 "schema": schema["schema"],
             },
         )
@@ -92,8 +108,9 @@ def get_tables_for_clone(self, schema):
             ):
                 continue
 
-            tables_for_clone[f"{self.config.env_prefix}{r['database_name']}.{r['schema_name']}.{r['name']}"] = {
-                "database": r["database_name"],
+            tables_for_clone[f"{schema['dst_database']}.{r['schema_name']}.{r['name']}"] = {
+                "src_database": schema["src_database"],
+                "dst_database": schema["dst_database"],
                 "schema": r["schema_name"],
                 "name": r["name"],
                 "is_transient": r["kind"] == "TRANSIENT",
@@ -119,27 +136,30 @@ def drop_object(self, row: dict):
             query.append("TRANSIENT")
 
         query.append(
-            "TABLE IF NOT EXISTS {database_with_prefix:i}.{schema:i}.{table_name:i}",
+            "TABLE IF NOT EXISTS {dst_database:i}.{schema:i}.{name:i}",
             {
-                "database_with_prefix": f"{self.config.env_prefix}{row['database']}",
+                "dst_database": row["dst_database"],
                 "schema": row["schema"],
-                "table_name": row["name"],
+                "name": row["name"],
             },
         )
 
         query.append_nl(
-            "CLONE {database:i}.{schema:i}.{table_name:i}",
+            "CLONE {src_database:i}.{schema:i}.{name:i}",
             {
-                "database": row["database"],
+                "src_database": row["src_database"],
                 "schema": row["schema"],
-                "table_name": row["name"],
+                "name": row["name"],
             },
         )
 
         cur = self.engine.execute_clone(query)
         r = cur.fetchone()
 
         if str(r["status"]).endswith("successfully created."):
+            self._drop_existing_policy_refs(
+                ObjectType.TABLE, SchemaObjectIdent("", row["dst_database"], row["schema"], row["name"])
+            )
             return ResolveResult.CREATE
 
         return ResolveResult.NOCHANGE
@@ -155,3 +175,59 @@ def _is_skipped(self):
             return True
 
         return False
+
+    def _drop_existing_policy_refs(self, object_type: ObjectType, object_name: SchemaObjectIdent):
+        cur = self.engine.execute_meta(
+            "SELECT * FROM TABLE(snowflake.information_schema.policy_references(ref_entity_domain => {object_type}, ref_entity_name => {object_name}))",
+            {
+                "object_type": object_type.singular_for_ref,
+                "object_name": object_name,
+            },
+        )
+
+        for r in cur:
+            if r["POLICY_KIND"] == "AGGREGATION_POLICY":
+                self.engine.execute_clone(
+                    "ALTER {object_type:r} {object_name:i} UNSET AGGREGATION POLICY",
+                    {
+                        "object_type": object_type.singular_for_ref,
+                        "object_name": object_name,
+                    },
+                )
+
+            elif r["POLICY_KIND"] == "MASKING_POLICY":
+                self.engine.execute_clone(
+                    "ALTER {object_type:r} {object_name:i} MODIFY COLUMN {column:i} UNSET MASKING POLICY",
+                    {
+                        "object_type": object_type.singular_for_ref,
+                        "object_name": object_name,
+                        "column": r["REF_COLUMN_NAME"],
+                    },
+                )
+
+            elif r["POLICY_KIND"] == "PROJECTION_POLICY":
+                self.engine.execute_clone(
+                    "ALTER {object_type:r} {object_name:i} MODIFY COLUMN {column:i} UNSET PROJECTION POLICY",
+                    {
+                        "object_type": object_type.singular_for_ref,
+                        "object_name": object_name,
+                        "column": r["REF_COLUMN_NAME"],
+                    },
+                )
+
+            elif r["POLICY_KIND"] == "ROW_ACCESS_POLICY":
+                policy_name = SchemaObjectIdent("", r["POLICY_DB"], r["POLICY_SCHEMA"], r["POLICY_NAME"])
+
+                self.engine.execute_clone(
+                    "ALTER {object_type:r} {object_name:i} DROP ROW ACCESS POLICY {policy_name:i}",
+                    {
+                        "object_type": object_type.singular_for_ref,
+                        "object_name": object_name,
+                        "policy_name": policy_name,
+                    },
+                )
+
+            else:
+                self.engine.logger.warning(
+                    f"Detected unknown policy type [{r['POLICY_KIND']} attached to cloned object [{object_name}]"
+                )

snowddl/settings.py

@@ -24,6 +24,7 @@ class SnowDDLSettings(BaseModelWithConfig):
     refresh_stage_encryption: bool = False
     refresh_secrets: bool = False
     clone_table: bool = False
+    clone_source_env_prefix: str = ""
     exclude_object_types: List[ObjectType] = []
     include_object_types: List[ObjectType] = []
     include_databases: List[DatabaseIdent] = []

snowddl/version.py

@@ -1 +1 @@
-__version__ = "0.52.0"
+__version__ = "0.53.0"