Code review comments.

dbt/adapters/snowflake/impl.py

@@ -22,7 +22,7 @@
 
 from dbt.adapters.snowflake.relation_configs import (
     SnowflakeRelationType,
-    SnowflakeObjectFormat,
+    TableFormat,
 )
 from dbt.adapters.snowflake import SnowflakeColumn
 from dbt.adapters.snowflake import SnowflakeConnectionManager
@@ -32,7 +32,6 @@
     import agate
 
 SHOW_OBJECT_METADATA_MACRO_NAME = "snowflake__show_object_metadata"
-LIST_ICEBERG_RELATIONS_MACRO_NAME = "snowflake__show_iceberg_relations"
 
 
 @dataclass
@@ -270,9 +269,7 @@ def _parse_list_relations_result(self, result: "agate.Row") -> SnowflakeRelation
             relation_type = self.Relation.DynamicTable
 
         table_format: str = (
-            SnowflakeObjectFormat.ICEBERG
-            if is_iceberg in ("Y", "YES")
-            else SnowflakeObjectFormat.DEFAULT
+            TableFormat.ICEBERG if is_iceberg in ("Y", "YES") else TableFormat.DEFAULT
         )
         quote_policy = {"database": True, "schema": True, "identifier": True}
 

dbt/adapters/snowflake/relation.py

@@ -1,28 +1,28 @@
 import textwrap
 
 from dataclasses import dataclass, field
-from typing import FrozenSet, Optional, Type
+from typing import FrozenSet, Optional, Type, Self, Iterator
 
 
 from dbt.adapters.base.relation import BaseRelation
 from dbt.adapters.contracts.relation import ComponentName, RelationConfig
-from dbt.adapters.events.types import AdapterEventWarning
+from dbt.adapters.events.types import AdapterEventWarning, AdapterEventDebug
 from dbt.adapters.relation_configs import (
     RelationConfigBase,
     RelationConfigChangeAction,
     RelationResults,
 )
 from dbt.adapters.utils import classproperty
 from dbt_common.exceptions import DbtRuntimeError
-from dbt_common.events.functions import warn_or_error
+from dbt_common.events.functions import fire_event, warn_or_error
 
 from dbt.adapters.snowflake.relation_configs import (
     SnowflakeDynamicTableConfig,
     SnowflakeDynamicTableConfigChangeset,
     SnowflakeDynamicTableRefreshModeConfigChange,
     SnowflakeDynamicTableTargetLagConfigChange,
     SnowflakeDynamicTableWarehouseConfigChange,
-    SnowflakeObjectFormat,
+    TableFormat,
     SnowflakeQuotePolicy,
     SnowflakeRelationType,
 )
@@ -31,7 +31,7 @@
 @dataclass(frozen=True, eq=False, repr=False)
 class SnowflakeRelation(BaseRelation):
     type: Optional[SnowflakeRelationType] = None
-    table_format: str = SnowflakeObjectFormat.DEFAULT
+    table_format: str = TableFormat.DEFAULT
     quote_policy: SnowflakeQuotePolicy = field(default_factory=lambda: SnowflakeQuotePolicy())
     require_alias: bool = False
     relation_configs = {
@@ -62,7 +62,7 @@ def is_dynamic_table(self) -> bool:
 
     @property
     def is_iceberg_format(self) -> bool:
-        return self.table_format == SnowflakeObjectFormat.ICEBERG
+        return self.table_format == TableFormat.ICEBERG
 
     @classproperty
     def DynamicTable(cls) -> str:
@@ -132,7 +132,7 @@ def as_case_sensitive(self) -> "SnowflakeRelation":
 
         return self.replace_path(**path_part_map)
 
-    def get_ddl_prefix_for_create(self, config: RelationConfig, temporary: bool):
+    def get_ddl_prefix_for_create(self, config: RelationConfig, temporary: bool) -> str:
         """
         This macro renders the appropriate DDL prefix during the create_table_as
         macro. It decides based on mutually exclusive table configuration options:
@@ -177,14 +177,14 @@ def get_ddl_prefix_for_create(self, config: RelationConfig, temporary: bool):
         else:
             return ""
 
-    def get_ddl_prefix_for_alter(self):
+    def get_ddl_prefix_for_alter(self) -> str:
         """All ALTER statements on Iceberg tables require an ICEBERG prefix"""
         if self.is_iceberg_format:
             return "iceberg"
         else:
             return ""
 
-    def render_iceberg_ddl(self, config: RelationConfig):
+    def get_iceberg_ddl_options(self, config: RelationConfig) -> str:
         base_location: str = f"_dbt/{self.schema}/{self.name}"
 
         if subpath := config.get("base_location_subpath"):
@@ -196,3 +196,57 @@ def render_iceberg_ddl(self, config: RelationConfig):
         base_location = '{base_location}'
         """
         return textwrap.indent(textwrap.dedent(iceberg_ddl_predicates), " " * 10)
+
+    def __drop_conditions(self, old_relation: Self) -> Iterator[tuple[bool, str]]:
+        drop_view_message: str = (
+            f"Dropping relation {old_relation} because it is a view and target relation {self} "
+            f"is of type {self.type}."
+        )
+
+        drop_table_for_iceberg_message: str = (
+            f"Dropping relation {old_relation} because it is a default format table "
+            f"and target relation {self} is an Iceberg format table."
+        )
+
+        drop_iceberg_for_table_message: str = (
+            f"Dropping relation {old_relation} because it is an Iceberg format table "
+            f"and target relation {self} is a default format table."
+        )
+
+        # An existing view must be dropped for model to build into a table".
+        yield (not old_relation.is_table, drop_view_message)
+        # An existing table must be dropped for model to build into an Iceberg table.
+        yield (
+            old_relation.is_table
+            and not old_relation.is_iceberg_format
+            and self.is_iceberg_format,
+            drop_table_for_iceberg_message,
+        )
+        # existing Iceberg table must be dropped for model to build into a table.
+        yield (
+            old_relation.is_table
+            and old_relation.is_iceberg_format
+            and not self.is_iceberg_format,
+            drop_iceberg_for_table_message,
+        )
+
+    def needs_to_drop(self, old_relation: Optional[Self]) -> bool:
+        """
+        To convert between Iceberg and non-Iceberg relations, a preemptive drop is
+        required.
+
+        drops cause latency, but it should be a relatively infrequent occurrence.
+
+        Some Boolean expression below are logically redundant, but this is done for easier
+        readability.
+        """
+
+        if old_relation is None:
+            return False
+
+        for condition, message in self.__drop_conditions(old_relation):
+            if condition:
+                fire_event(AdapterEventDebug(base_msg=message))
+                return True
+
+        return False

dbt/adapters/snowflake/relation_configs/__init__.py

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

dbt/adapters/snowflake/relation_configs/formats.py

@@ -1,7 +1,12 @@
 from dbt_common.dataclass_schema import StrEnum  # doesn't exist in standard library until py3.11
 
 
-class SnowflakeObjectFormat(StrEnum):
+class TableFormat(StrEnum):
+    """
+    Snowflake docs refers to this an 'Object Format.'
+    Data practitioners and interfaces refer to this as 'Table Format's, hence the term's use here.
+    """
+
     DEFAULT = "default"
     ICEBERG = "iceberg"
 

dbt/include/snowflake/macros/adapters.sql

@@ -137,27 +137,19 @@
 {% macro snowflake__list_relations_without_caching(schema_relation, max_iter=10, max_results_per_iter=10000) %}
 
   {%- set max_total_results = max_results_per_iter * max_iter -%}
-  {% if schema_relation is string %}
-    {%- set sql -%}
+  {%- set sql -%}
+    {% if schema_relation is string %}
       show objects in {{ schema_relation }} limit {{ max_results_per_iter }};
-      select all_objects.*, is_iceberg as "is_iceberg"
-      from table(result_scan(last_query_id(-1))) all_objects
-      left join INFORMATION_SCHEMA.tables as all_tables
-        on all_tables.table_name = all_objects."name"
-        and all_tables.table_schema = all_objects."schema_name"
-        and all_tables.table_catalog = all_objects."database_name"
-    {%- endset -%}
-  {% else %}
-    {%- set sql -%}
+    {% else %}
       show objects in {{ schema_relation.include(identifier=False) }} limit {{ max_results_per_iter }};
+    {% endif -%}
       select all_objects.*, is_iceberg as "is_iceberg"
       from table(result_scan(last_query_id(-1))) all_objects
       left join INFORMATION_SCHEMA.tables as all_tables
         on all_tables.table_name = all_objects."name"
         and all_tables.table_schema = all_objects."schema_name"
         and all_tables.table_catalog = all_objects."database_name"
-    {%- endset -%}
-  {% endif -%}
+  {%- endset -%}
 
   {%- set result = run_query(sql) -%}
 

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

@@ -18,7 +18,9 @@
 
   {{ run_hooks(pre_hooks) }}
 
-  {{ drop_old_relation_as_needed(old_relation, target_relation) }}
+  {% if target_relation.needs_to_drop(old_relation) %}
+    {{ drop_relation_if_exists(old_relation) }}
+  {% endif %}
 
   {% call statement('main', language=language) -%}
       {{ create_table_as(False, target_relation, compiled_code, language) }}
@@ -84,34 +86,3 @@ 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

@@ -26,8 +26,7 @@
             Valid DDL in CTAS statements. Plain create statements have a different order.
             https://docs.snowflake.com/en/sql-reference/sql/create-iceberg-table
           #}
-          {{ relation.render_iceberg_ddl(config.model.config) }}
-        {% else %}
+          {{ relation.get_iceberg_ddl_options(config.model.config) }}
         {%- endif -%}
 
         {%- set contract_config = config.get('contract') -%}

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.path.schema)
+            relations = my_adapter.list_relations_without_caching(schema)
         return relations
 
     def test_list_relations_without_caching(self, project):