support on_table_exists=skip

implement table materialization option on_table_exists=skip to allow
user to skip a model if the target relation already exists.

Author: choyrim

.changes/unreleased/Features-20250602-212247.yaml

@@ -0,0 +1,7 @@
+kind: Features
+body: table materialization on_table_exists=skip option
+time: 2025-06-02T21:22:47.837474-04:00
+custom:
+    Author: choyrim
+    Issue: "479"
+    PR: "481"

dbt/include/trino/macros/adapters.sql

@@ -145,18 +145,15 @@
   {%- endif -%}
 {%- endmacro -%}
 
-{% macro trino__create_table_as(temporary, relation, sql, replace=False) -%}
+{% macro trino__create_table_as(temporary, relation, sql, on_exists=None) -%}
 
-  {%- if replace -%}
-    {%- set or_replace = ' or replace' -%}
-  {%- else -%}
-    {%- set or_replace = '' -%}
-  {%- endif -%}
+  {%- set or_replace = ' or replace' if on_exists == 'replace' else '' -%}
+  {%- set if_not_exists = ' if not exists' if on_exists == 'skip' else '' -%}
 
   {%- set contract_config = config.get('contract') -%}
   {%- if contract_config.enforced -%}
 
-  create{{ or_replace }} table
+  create{{ or_replace }} table{{ if_not_exists }}
     {{ relation }}
     {{ get_table_columns_and_constraints() }}
     {{ get_assert_columns_equivalent(sql) }}
@@ -173,14 +170,14 @@
 
   {%- else %}
 
-    create{{ or_replace }} table {{ relation }}
+    create{{ or_replace }} table{{ if_not_exists }} {{ relation }}
       {{ comment(model.get('description')) }}
       {{ properties(temporary) }}
     as (
       {{ sql }}
     );
 
-   {%- endif %}
+  {%- endif %}
 {% endmacro %}
 
 

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

@@ -1,6 +1,6 @@
 {% materialization table, adapter = 'trino' %}
   {%- set on_table_exists = config.get('on_table_exists', 'rename') -%}
-  {% if on_table_exists not in ['rename', 'drop', 'replace'] %}
+  {% if on_table_exists not in ['rename', 'drop', 'replace', 'skip'] %}
       {%- set log_message = 'Invalid value for on_table_exists (%s) specified. Setting default value (%s).' % (on_table_exists, 'rename') -%}
       {% do log(log_message) %}
       {%- set on_table_exists = 'rename' -%}
@@ -84,7 +84,14 @@
   {% elif on_table_exists == 'replace' %}
       {#-- build model #}
       {% call statement('main') -%}
-        {{ create_table_as(False, target_relation, sql, True) }}
+        {{ create_table_as(False, target_relation, sql, 'replace') }}
       {%- endcall %}
+
+  {% elif on_table_exists == 'skip' %}
+      {#-- build model #}
+      {% call statement('main') -%}
+        {{ create_table_as(False, target_relation, sql, 'skip') }}
+      {%- endcall %}
+
   {% endif %}
 {% endmacro %}

tests/functional/adapter/materialization/test_on_table_exists.py

@@ -309,3 +309,52 @@ class TestOnTableExistsReplaceDeltaIncrementalFullRefresh(
     BaseOnTableExistsReplaceIncrementalFullRefresh
 ):
     pass
+
+
+class TestOnTableExistsSkip(BaseOnTableExists):
+    """
+    Testing on_table_exists = `skip` configuration for table materialization,
+    using dbt seed, run and tests commands and validate data load correctness.
+    """
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "name": "table_rename",
+            "models": {"+materialized": "table", "+on_table_exists": "skip"},
+            "seeds": {
+                "+column_types": {"some_date": "timestamp(6)"},
+            },
+        }
+
+    # The actual sequence of dbt commands and assertions
+    # pytest will take care of all "setup" + "teardown"
+    def test_run_seed_test(self, project):
+        # seed seeds
+        results = run_dbt(["seed"], expect_pass=True)
+        assert len(results) == 1
+        # run models two times to check on_table_exists = 'skip'
+        results, logs = run_dbt_and_capture(["--debug", "run"], expect_pass=True)
+        assert len(results) == 1
+        assert (
+            f'create table if not exists "{project.database}"."{project.test_schema}"."materialization"'
+            in logs
+        )
+        assert "alter table" not in logs
+        assert "drop table" not in logs
+        assert "or replace" not in logs
+        results, logs = run_dbt_and_capture(["--debug", "run"], expect_pass=True)
+        assert len(results) == 1
+        assert (
+            f'create table if not exists "{project.database}"."{project.test_schema}"."materialization"'
+            in logs
+        )
+        assert "alter table" not in logs
+        assert "drop table" not in logs
+        assert "or replace" not in logs
+        # test tests
+        results = run_dbt(["test"], expect_pass=True)
+        assert len(results) == 3
+
+        # check if the data was loaded correctly
+        check_relations_equal(project.adapter, ["seed", "materialization"])