resolve rebasing artifacts

Author: mikealfare

dbt-adapters/.changes/unreleased/Features-20250219-093958.yaml

@@ -8,11 +8,19 @@
 from dbt.adapters.snowflake.catalogs._iceberg_rest import (
     IcebergAWSGlueCatalogIntegration,
     IcebergRESTCatalogIntegration,
+    IcebergRESTCatalogRelation,
 )
 
 
-SnowflakeCatalogRelation = Union[IcebergManagedCatalogRelation]
-SnowflakeCatalogIntegration = Union[IcebergManagedCatalogIntegration]
+SnowflakeCatalogRelation = Union[
+    IcebergManagedCatalogRelation,
+    IcebergRESTCatalogRelation,
+]
+SnowflakeCatalogIntegration = Union[
+    IcebergAWSGlueCatalogIntegration,
+    IcebergManagedCatalogIntegration,
+    IcebergRESTCatalogIntegration,
+]
 
 
 CATALOG_INTEGRATIONS = [

dbt-snowflake/src/dbt/adapters/snowflake/catalogs/__init__.py

@@ -1,13 +1,16 @@
 from dataclasses import dataclass, field
-from typing import Any, Dict, Optional, TYPE_CHECKING
-from typing_extensions import Self
+from typing import Any, Dict, Optional
 
 from dbt.adapters.catalogs import CatalogIntegration, CatalogIntegrationConfig
-from dbt.adapters.relation_configs import RelationResults
-from dbt_common.exceptions import DbtInternalError
+from dbt.adapters.contracts.relation import RelationConfig
 
-if TYPE_CHECKING:
-    import agate
+
+@dataclass
+class IcebergRESTCatalogRelation:
+    base_location: str
+    catalog_name: str
+    external_volume: Optional[str] = None
+    table_format: str = "iceberg"
 
 
 @dataclass
@@ -60,36 +63,34 @@ class IcebergRESTCatalogIntegration(CatalogIntegration):
         -   must be False
     """
 
+    catalog_type: str = "iceberg_rest"
     table_format: str = "iceberg"
     allows_writes: bool = False
 
     def __init__(self, config: CatalogIntegrationConfig) -> None:
         super().__init__(config)
-        if self.catalog_type not in ["iceberg_rest", "aws_glue"]:
-            raise DbtInternalError(
-                f"Attempting to create IcebergREST catalog integration for catalog {self.name} with catalog type {config.catalog_type}."
-            )
-        if self.table_format and self.table_format != "iceberg":
-            raise DbtInternalError(
-                f"Unsupported table format for catalog {self.name}: {self.table_format}. Expected `iceberg` or unset."
-            )
-
         if config.adapter_properties:
             self.namespace = config.adapter_properties.get("namespace")
 
-    @classmethod
-    def from_relation_results(cls, relation_results: RelationResults) -> Self:
-        table: "agate.Row" = relation_results["table"][0]
-        catalog: "agate.Row" = relation_results["catalog"][0]
+    def build_relation(self, config: RelationConfig) -> IcebergRESTCatalogRelation:
+        return IcebergRESTCatalogRelation(
+            base_location=self.__base_location(config),
+            external_volume=config.config.extra.get("external_volume", self.external_volume),
+            catalog_name=self.catalog_name,
+        )
+
+    @staticmethod
+    def __base_location(config: RelationConfig) -> str:
+        # If the base_location_root config is supplied, overwrite the default value ("_dbt/")
+        prefix = config.config.extra.get("base_location_root", "_dbt")
 
-        adapter_properties = {}
-        if namespace := catalog.get("namespace"):
-            adapter_properties["namespace"] = namespace
+        base_location = f"{prefix}/{config.schema}/{config.identifier}"
 
-        config = IcebergRESTCatalogIntegrationConfig(
-            name=catalog.get("catalog_name"),
-            catalog_type=catalog.get("catalog_type"),
-            external_volume=table.get("external_volume_name"),
-            adapter_properties=adapter_properties,
-        )
-        return cls(config)
+        if subpath := config.config.extra.get("base_location_subpath"):
+            base_location += f"/{subpath}"
+
+        return base_location
+
+
+class IcebergAWSGlueCatalogIntegration(IcebergRESTCatalogIntegration):
+    catalog_type: str = "aws_glue"

dbt-snowflake/src/dbt/adapters/snowflake/catalogs/_iceberg_rest.py

@@ -6,7 +6,6 @@
 from dbt.adapters.capability import CapabilityDict, CapabilitySupport, Support, Capability
 from dbt.adapters.catalogs import CatalogRelation
 from dbt.adapters.contracts.relation import RelationConfig
-from dbt.adapters.snowflake.catalogs import CATALOG_INTEGRATIONS
 from dbt.adapters.sql import SQLAdapter
 from dbt.adapters.sql.impl import (
     LIST_SCHEMAS_MACRO_NAME,
@@ -62,7 +61,6 @@ class SnowflakeConfig(AdapterConfig):
     external_volume: Optional[str] = None
     base_location_root: Optional[str] = None
     base_location_subpath: Optional[str] = None
-    catalog_name: Optional[str] = None
 
 
 class SnowflakeAdapter(SQLAdapter):
@@ -71,7 +69,8 @@ class SnowflakeAdapter(SQLAdapter):
     ConnectionManager = SnowflakeConnectionManager
 
     AdapterSpecificConfigs = SnowflakeConfig
-    CATALOG_INTEGRATIONS = CATALOG_INTEGRATIONS  # type:ignore
+
+    CATALOG_INTEGRATIONS = CATALOG_INTEGRATIONS
     CONSTRAINT_SUPPORT = {
         ConstraintType.check: ConstraintSupport.NOT_SUPPORTED,
         ConstraintType.not_null: ConstraintSupport.ENFORCED,

dbt-snowflake/src/dbt/adapters/snowflake/impl.py

@@ -10,10 +10,6 @@
     {%- set transient='' -%}
 {%- endif -%}
 
-{%- set sql_header = config.get('sql_header', none) -%}
-
-{%- set copy_grants = config.get('copy_grants', default=false) -%}
-
 {%- set cluster_by_keys = config.get('cluster_by', default=none) -%}
 {%- set enable_automatic_clustering = config.get('automatic_clustering', default=false) -%}
 {%- set copy_grants = config.get('copy_grants', default=false) -%}