Add external_volume and catalog parameters to DATABASE object type, update tests

Author: littleK0i

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## [0.63.1] - 2026-03-02
+
+- Added `external_volume` and `catalog` parameters to `DATABASE` object type in order to support Iceberg tables on database level.
+
 ## [0.63.0] - 2026-02-04
 
 - Introduced support for `UUID` data type.

snowddl/blueprint/blueprint.py

@@ -130,6 +130,8 @@ class DatabaseBlueprint(AbstractBlueprint):
     is_sandbox: Optional[bool] = None
     is_transient: Optional[bool] = None
     retention_time: Optional[int] = None
+    external_volume: Optional[Ident] = None
+    catalog: Optional[Ident] = None
     owner_database_write: List[IdentPattern] = []
     owner_database_read: List[IdentPattern] = []
     owner_schema_write: List[IdentPattern] = []

snowddl/parser/database.py

@@ -23,6 +23,12 @@
         "retention_time": {
             "type": "integer"
         },
+        "external_volume": {
+            "type": "string"
+        },
+        "catalog": {
+            "type": "string",
+        },
         "is_sandbox": {
             "type": "boolean"
         },
@@ -103,6 +109,8 @@ def load_blueprints(self):
                 permission_model=database_permission_model_name,
                 is_transient=database_params.get("is_transient", False),
                 retention_time=database_params.get("retention_time", None),
+                external_volume=Ident(database_params.get("external_volume")) if database_params.get("external_volume") else None,
+                catalog=Ident(database_params.get("catalog")) if database_params.get("catalog") else None,
                 is_sandbox=database_params.get("is_sandbox", False),
                 owner_database_write=[IdentPattern(p) for p in database_params.get("owner_database_write", [])],
                 owner_database_read=[IdentPattern(p) for p in database_params.get("owner_database_read", [])],

snowddl/parser/iceberg_table.py

@@ -2,6 +2,7 @@
 
 from snowddl.blueprint import IcebergTableBlueprint, Ident, SchemaObjectIdent
 from snowddl.parser.abc_parser import AbstractParser, ParsedFile
+from snowddl.parser.database import database_json_schema
 from snowddl.parser.schema import schema_json_schema
 
 
@@ -46,22 +47,30 @@ def load_blueprints(self):
         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] = schema_params
+
+                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"),
+                    "is_sandbox": schema_params.get("is_sandbox", database_params.get("is_sandbox", False)),
+                    "external_volume": schema_params.get("external_volume", database_params.get("external_volume")),
+                    "catalog": schema_params.get("catalog", database_params.get("catalog")),
+                }
 
         self.parse_schema_object_files(
             "iceberg_table", iceberg_table_json_schema, partial(self.process_table, combined_params=combined_params)
         )
 
     def process_table(self, f: ParsedFile, combined_params: dict):
         if not combined_params[f.database][f.schema].get("external_volume"):
-            raise ValueError("Iceberg table requires parameter [external_volume] to be defined on schema level")
+            raise ValueError("Iceberg table requires parameter [external_volume] to be defined on database or schema level")
 
         if not combined_params[f.database][f.schema].get("catalog"):
-            raise ValueError("Iceberg table requires parameter [catalog] to be defined on schema level")
+            raise ValueError("Iceberg table requires parameter [catalog] to be defined on database or schema level")
 
         bp = IcebergTableBlueprint(
             full_name=SchemaObjectIdent(self.env_prefix, f.database, f.schema, f.name),

snowddl/parser/schema.py

@@ -112,6 +112,8 @@ def load_blueprints(self):
                     "is_transient": database_params.get("is_transient", False) or schema_params.get("is_transient", False),
                     "retention_time": schema_params.get("retention_time"),
                     "is_sandbox": schema_params.get("is_sandbox", database_params.get("is_sandbox", False)),
+                    "external_volume": schema_params.get("external_volume", database_params.get("external_volume")),
+                    "catalog": schema_params.get("catalog", database_params.get("catalog")),
                 }
 
                 # fmt: off
@@ -121,8 +123,8 @@ def load_blueprints(self):
                     is_sandbox=combined_params.get("is_sandbox", False),
                     is_transient=combined_params.get("is_transient", False),
                     retention_time=combined_params.get("retention_time", None),
-                    external_volume=Ident(schema_params.get("external_volume")) if schema_params.get("external_volume") else None,
-                    catalog=Ident(schema_params.get("catalog")) if schema_params.get("catalog") else None,
+                    external_volume=Ident(combined_params.get("external_volume")) if combined_params.get("external_volume") else None,
+                    catalog=Ident(combined_params.get("catalog")) if combined_params.get("catalog") else None,
                     owner_database_write=[IdentPattern(p) for p in schema_params.get("owner_database_write", [])],
                     owner_database_read=[IdentPattern(p) for p in schema_params.get("owner_database_read", [])],
                     owner_schema_write=[IdentPattern(p) for p in schema_params.get("owner_schema_write", [])],

snowddl/resolver/database_owner_role.py

@@ -59,6 +59,25 @@ def get_blueprint_owner_role(self, database_bp: DatabaseBlueprint):
             )
         )
 
+        # Iceberg-related grants
+        if database_bp.external_volume:
+            grants.append(
+                Grant(
+                    privilege="USAGE",
+                    on=ObjectType.VOLUME,
+                    name=database_bp.external_volume,
+                )
+            )
+
+        if database_bp.catalog:
+            grants.append(
+                Grant(
+                    privilege="USAGE",
+                    on=ObjectType.INTEGRATION,
+                    name=database_bp.catalog,
+                )
+            )
+
         # Create grants
         for model_create_grant in database_permission_model.owner_create_grants:
             future_grants.append(

snowddl/version.py

@@ -1 +1 @@
-__version__ = "0.63.0"
+__version__ = "0.63.1"

test/_config/step1/iceberg_db1/iceberg_sc1/params.yaml

@@ -1,4 +1,4 @@
-permission_model: iceberg
+permission_model: iceberg_schema_owner_model
 
 external_volume: test_external_volume_glue
 catalog: test_catalog_glue

test/_config/step1/iceberg_db1/iceberg_sc2/params.yaml

@@ -1,4 +1,4 @@
-permission_model: iceberg
+permission_model: iceberg_schema_owner_model
 
 external_volume: test_external_volume_glue
 catalog: test_catalog_object_store

test/_config/step1/iceberg_db2/iceberg_sc3/iceberg_table/it003_it1.yaml

@@ -0,0 +1,2 @@
+catalog_table_name: test_iceberg_table_1
+comment: abc