Add support for Iceberg Table Materialization

dbt/adapters/snowflake/impl.py

@@ -20,7 +20,10 @@
 from dbt_common.exceptions import CompilationError, DbtDatabaseError, DbtRuntimeError
 from dbt_common.utils import filter_null_values
 
-from dbt.adapters.snowflake.relation_configs import SnowflakeRelationType
+from dbt.adapters.snowflake.relation_configs import (
+    SnowflakeRelationType,
+    SnowflakeObjectFormat,
+)
 from dbt.adapters.snowflake import SnowflakeColumn
 from dbt.adapters.snowflake import SnowflakeConnectionManager
 from dbt.adapters.snowflake import SnowflakeRelation
@@ -29,6 +32,7 @@
     import agate
 
 SHOW_OBJECT_METADATA_MACRO_NAME = "snowflake__show_object_metadata"
+LIST_ICEBERG_RELATIONS_MACRO_NAME = "snowflake__show_iceberg_relations"
 
 
 @dataclass
@@ -44,6 +48,11 @@ class SnowflakeConfig(AdapterConfig):
     merge_update_columns: Optional[str] = None
     target_lag: Optional[str] = None
 
+    # extended formats
+    object_format: Optional[str] = None
+    external_volume: Optional[str] = None
+    base_location_subpath: Optional[str] = None
+
 
 class SnowflakeAdapter(SQLAdapter):
     Relation = SnowflakeRelation
@@ -223,8 +232,35 @@ def list_relations_without_caching(
         self, schema_relation: SnowflakeRelation
     ) -> List[SnowflakeRelation]:
         kwargs = {"schema_relation": schema_relation}
+
+        def check_is_iceberg(row, table2):
+            for match_row in table2.rows:
+                if (
+                    row["name"] == match_row["name"]
+                    and row["database_name"] == match_row["database_name"]
+                    and row["schema_name"] == match_row["schema_name"]
+                ):
+                    return "Y"
+            return "N"
+
         try:
-            results = self.execute_macro(LIST_RELATIONS_MACRO_NAME, kwargs=kwargs)
+            schema_objects = self.execute_macro(LIST_RELATIONS_MACRO_NAME, kwargs=kwargs)
+            iceberg_table_results = self.execute_macro(
+                LIST_ICEBERG_RELATIONS_MACRO_NAME, kwargs=kwargs
+            )
+            import agate
+
+            # this only seems to only inflate runtime 16%; TODO: stress test
+            results = schema_objects.compute(
+                [
+                    (
+                        "is_iceberg",
+                        agate.Formula(
+                            agate.Text(), lambda row: check_is_iceberg(row, iceberg_table_results)
+                        ),
+                    )
+                ]
+            )
         except DbtDatabaseError as exc:
             # if the schema doesn't exist, we just want to return.
             # Alternatively, we could query the list of schemas before we start
@@ -237,16 +273,19 @@ def list_relations_without_caching(
         columns = ["database_name", "schema_name", "name", "kind"]
         if "is_dynamic" in results.column_names:
             columns.append("is_dynamic")
+        if "is_iceberg" in results.column_names:
+            columns.append("is_iceberg")
 
         return [self._parse_list_relations_result(result) for result in results.select(columns)]
 
     def _parse_list_relations_result(self, result: "agate.Row") -> SnowflakeRelation:
         # this can be reduced to always including `is_dynamic` once bundle `2024_03` is mandatory
         try:
-            database, schema, identifier, relation_type, is_dynamic = result
+            database, schema, identifier, relation_type, is_dynamic, is_iceberg = result
         except ValueError:
             database, schema, identifier, relation_type = result
             is_dynamic = "N"
+            is_iceberg = "N"
 
         try:
             relation_type = self.Relation.get_relation_type(relation_type.lower())
@@ -256,12 +295,16 @@ def _parse_list_relations_result(self, result: "agate.Row") -> SnowflakeRelation
         if relation_type == self.Relation.Table and is_dynamic == "Y":
             relation_type = self.Relation.DynamicTable
 
+        object_format: str = (
+            SnowflakeObjectFormat.ICEBERG if is_iceberg == "Y" else SnowflakeObjectFormat.DEFAULT
+        )
         quote_policy = {"database": True, "schema": True, "identifier": True}
         return self.Relation.create(
             database=database,
             schema=schema,
             identifier=identifier,
             type=relation_type,
+            object_format=object_format,
             quote_policy=quote_policy,
         )
 

dbt/adapters/snowflake/relation.py

@@ -17,6 +17,7 @@
     SnowflakeDynamicTableRefreshModeConfigChange,
     SnowflakeDynamicTableTargetLagConfigChange,
     SnowflakeDynamicTableWarehouseConfigChange,
+    SnowflakeObjectFormat,
     SnowflakeQuotePolicy,
     SnowflakeRelationType,
 )
@@ -25,6 +26,7 @@
 @dataclass(frozen=True, eq=False, repr=False)
 class SnowflakeRelation(BaseRelation):
     type: Optional[SnowflakeRelationType] = None
+    object_format: str = SnowflakeObjectFormat.DEFAULT
     quote_policy: SnowflakeQuotePolicy = field(default_factory=lambda: SnowflakeQuotePolicy())
     require_alias: bool = False
     relation_configs = {
@@ -53,6 +55,10 @@ class SnowflakeRelation(BaseRelation):
     def is_dynamic_table(self) -> bool:
         return self.type == SnowflakeRelationType.DynamicTable
 
+    @property
+    def is_iceberg_format(self) -> bool:
+        return self.object_format == SnowflakeObjectFormat.ICEBERG
+
     @classproperty
     def DynamicTable(cls) -> str:
         return str(SnowflakeRelationType.DynamicTable)

dbt/adapters/snowflake/relation_configs/__init__.py

@@ -10,3 +10,4 @@
     SnowflakeQuotePolicy,
     SnowflakeRelationType,
 )
+from dbt.adapters.snowflake.relation_configs.formats import SnowflakeObjectFormat

dbt/adapters/snowflake/relation_configs/formats.py

@@ -0,0 +1,9 @@
+from dbt_common.dataclass_schema import StrEnum  # doesn't exist in standard library until py3.11
+
+
+class SnowflakeObjectFormat(StrEnum):
+    DEFAULT = "default"
+    ICEBERG = "iceberg"
+
+    def __str__(self):
+        return self.value

dbt/include/snowflake/macros/adapters.sql

@@ -318,3 +318,12 @@
     {{ snowflake_dml_explicit_transaction(truncate_dml) }}
   {%- endcall %}
 {% endmacro %}
+
+
+{% macro snowflake__show_iceberg_relations(schema_relation) %}
+  {%- set sql -%}
+    show iceberg tables in {{ schema_relation }}
+  {%- endset -%}
+  {%- set result = run_query(sql) -%}
+  {%- do return(result) -%}
+{% endmacro %}

dbt/include/snowflake/macros/materializations/table.sql

@@ -8,18 +8,17 @@
   {% set grant_config = config.get('grants') %}
 
   {%- set old_relation = adapter.get_relation(database=database, schema=schema, identifier=identifier) -%}
-  {%- set target_relation = api.Relation.create(identifier=identifier,
-                                                schema=schema,
-                                                database=database, type='table') -%}
+  {%- set target_relation = api.Relation.create(
+	identifier=identifier,
+	schema=schema,
+	database=database,
+	type='table',
+	object_format=config.get('object_format', 'default')
+    ) -%}
 
   {{ run_hooks(pre_hooks) }}
 
-  {#-- Drop the relation if it was a view to "convert" it in a table. This may lead to
-    -- downtime, but it should be a relatively infrequent occurrence  #}
-  {% if old_relation is not none and not old_relation.is_table %}
-    {{ log("Dropping relation " ~ old_relation ~ " because it is of type " ~ old_relation.type) }}
-    {{ drop_relation_if_exists(old_relation) }}
-  {% endif %}
+  {{ drop_old_relation_as_needed(old_relation, target_relation) }}
 
   {% call statement('main', language=language) -%}
       {{ create_table_as(False, target_relation, compiled_code, language) }}
@@ -85,3 +84,34 @@ def main(session):
 # dbt = dbtObj(session.table)
 # df = model(dbt, session)
 {%endmacro%}
+
+
+{% macro drop_old_relation_as_needed(old_relation, target_relation) %}
+  {% if old_relation is none %}
+      {{ return('') }}
+  {% endif %}
+
+  {#
+    -- Each of these will cause some latency, but it shoudl be a relatively infrequent occurrence.
+
+    -- An existing view must be dropped for model to "convert" into a table"
+  #}
+  {% if not old_relation.is_table %}
+    {{ log("Dropping relation " ~ old_relation ~ " because it is of type " ~ old_relation.type) }}
+    {{ drop_relation_if_exists(old_relation) }}
+
+  {#
+    -- An existing Iceberg table must be dropped for model to "convert" into a table.
+  #}
+  {% elif old_relation.is_iceberg_format and not target_relation.is_iceberg_format %}
+    {{ log("Dropping relation " ~ old_relation ~ " because it is an Iceberg format table and target relation " ~ target_relation ~ " is a default format table.") }}
+    {{ drop_relation_if_exists(old_relation) }}
+
+  {#
+    -- An existing table must be dropped for model to "convert" into an Iceberg table.
+  #}
+  {% elif old_relation.is_table and not old_relation.is_iceberg_format and target_relation.is_iceberg_format %}
+    {{ log("Dropping relation " ~ old_relation ~ " because it is a default format table and target relation is an Iceberg format table.") }}
+    {{ drop_relation_if_exists(old_relation) }}
+  {% endif %}
+{% endmacro %}

dbt/include/snowflake/macros/relations/table/create.sql

@@ -1,14 +1,8 @@
 {% macro snowflake__create_table_as(temporary, relation, compiled_code, language='sql') -%}
-  {%- set transient = config.get('transient', default=true) -%}
-
-  {% if temporary -%}
-    {%- set table_type = "temporary" -%}
-  {%- elif transient -%}
-    {%- set table_type = "transient" -%}
-  {%- else -%}
-    {%- set table_type = "" -%}
-  {%- endif %}
+  {%- set materialization_prefix = get_create_ddl_prefix(temporary) -%}
+  {%- set alter_prefix = get_alter_ddl_prefix() -%}
 
+  {# Generate DDL/DML #}
   {%- if language == 'sql' -%}
     {%- set cluster_by_keys = config.get('cluster_by', default=none) -%}
     {%- set enable_automatic_clustering = config.get('automatic_clustering', default=false) -%}
@@ -26,7 +20,16 @@
 
     {{ sql_header if sql_header is not none }}
 
-        create or replace {{ table_type }} table {{ relation }}
+        create or replace {{ materialization_prefix }} table {{ relation }}
+        {%- if _is_iceberg_relation() %}
+          {#
+            Valid DDL in CTAS statements. Plain create statements have a different order.
+            https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table
+          #}
+          {{ render_iceberg_ddl(relation) }}
+        {% else %}
+        {%- endif -%}
+
         {%- set contract_config = config.get('contract') -%}
         {%- if contract_config.enforced -%}
           {{ get_assert_columns_equivalent(sql) }}
@@ -44,16 +47,97 @@
           {%- endif %}
         );
       {% if cluster_by_string is not none and not temporary -%}
-        alter table {{relation}} cluster by ({{cluster_by_string}});
+        alter {{ alter_prefix }} table {{relation}} cluster by ({{cluster_by_string}});
       {%- endif -%}
-      {% if enable_automatic_clustering and cluster_by_string is not none and not temporary  -%}
-        alter table {{relation}} resume recluster;
+      {% if enable_automatic_clustering and cluster_by_string is not none and not temporary %}
+        alter {{ alter_prefix }} table {{relation}} resume recluster;
       {%- endif -%}
 
   {%- elif language == 'python' -%}
+    {%- if iceberg -%}
+      {% do exceptions.raise_compiler_error('Iceberg is incompatible with Python models. Please use a SQL model for the iceberg format.') %}
+    {%- endif %}
     {{ py_write_table(compiled_code=compiled_code, target_relation=relation, table_type=table_type) }}
   {%- else -%}
       {% do exceptions.raise_compiler_error("snowflake__create_table_as macro didn't get supported language, it got %s" % language) %}
   {%- endif -%}
 
 {% endmacro %}
+
+
+{#
+ # Helper Macros
+ #}
+
+{% macro get_create_ddl_prefix(temporary) %}
+  {#
+    This macro generates the appropriate DDL prefix for creating a table in Snowflake,
+    considering the mutually exclusive nature of certain table types:
+
+    - ICEBERG: A specific storage format that requires a distinct DDL layout.
+    - TEMPORARY: Indicates a table that exists only for the duration of the session.
+    - TRANSIENT: A type of table that is similar to a permanent table but without fail-safe.
+
+    Note: If ICEBERG is specified, transient=true throws a warning because ICEBERG
+      does not support transient tables.
+  #}
+
+  {%- set is_iceberg   = _is_iceberg_relation() -%}
+  {%- set is_temporary = temporary -%}
+
+  {%- if is_iceberg -%}
+      {# -- Check if user supplied a transient model config of True. #}
+      {%- if config.get('transient') == True -%}
+        {{ exceptions.warn("Iceberg format relations cannot be transient. Please remove either the transient or iceberg parameters from %s. If left unmodified, dbt will ignore 'transient'." % this) }}
+      {%- endif %}
+
+      {# -- Check if runtime is trying to create a Temporary Iceberg table. #}
+      {%- if is_temporary -%}
+	{{ exceptions.raise_compiler_error("Iceberg format relations cannot be temporary. Temporary is being inserted into an Iceberg format table while materializing %s." % this) }}
+      {%- endif %}
+
+    {{ return('iceberg') }}
+
+  {%- elif is_temporary -%}
+    {{ return('temporary') }}
+
+  {# -- Always supply transient on table create DDL unless user specifically sets transient to false. #}
+  {%- elif config.get('transient', default=true) -%}
+    {{ return('transient') }}
+
+  {%- else -%}
+    {{ return('') }}
+  {%- endif -%}
+{% endmacro %}
+
+
+{% macro get_alter_ddl_prefix() %}
+  {# All ALTER statements on Iceberg tables require an ICEBERG prefix #}
+  {%- if _get_relation_object_format() == 'iceberg' -%}
+    {{ return('iceberg') }}
+  {%- else -%}
+    {{ return('') }}
+  {%- endif -%}
+{% endmacro %}
+
+
+{% macro _get_relation_object_format() %}
+  {{ return(config.get('object_format', default='')) }}
+{% endmacro %}
+
+
+{% macro _is_iceberg_relation() %}
+  {{ return(_get_relation_object_format() == 'iceberg') }}
+{% endmacro %}
+
+
+{% macro render_iceberg_ddl(relation) -%}
+  {%- set external_volume = config.get('external_volume') -%}
+  {# S3 treats subpaths with or without a trailing '/' as functionally equivalent #}
+  {%- set subpath = config.get('base_location_subpath') -%}
+  {%- set base_location = '_dbt/' ~ relation.schema ~ '/' ~ relation.name ~ (('/' ~ subpath) if subpath else '') -%}
+
+  external_volume = '{{ external_volume }}'
+  catalog = 'snowflake'
+  base_location = '{{ base_location }}'
+{% endmacro %}

tests/functional/adapter/list_relations_tests/test_show_objects.py

@@ -73,7 +73,7 @@ def list_relations_without_caching(project) -> List[SnowflakeRelation]:
             database=project.database, schema=project.test_schema, identifier=""
         )
         with get_connection(my_adapter):
-            relations = my_adapter.list_relations_without_caching(schema)
+            relations = my_adapter.list_relations_without_caching(schema.path.schema)
         return relations
 
     def test_list_relations_without_caching(self, project):