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

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' -%}
@@ -86,5 +86,26 @@
       {% call statement('main') -%}
         {{ create_table_as(False, target_relation, sql, True) }}
       {%- endcall %}
+
+  {% elif on_table_exists == 'skip' %}
+
+      {#-- table does not exists #}
+      {% if existing_relation is none %}
+        {#-- build model #}
+        {% call statement('main') -%}
+          {{ create_table_as(False, target_relation, sql) }}
+        {%- endcall %}
+
+      {#-- table does exists #}
+      {% else %}
+        {#-- do nothing #}
+        {%- set log_message = 'table %s already exists. skipping.' % (existing_relation) -%}
+        {% do log(log_message) %}
+        {% call statement('main') -%}
+          select 1
+        {%- endcall %}
+
+      {% endif %}
+
   {% endif %}
 {% endmacro %}

tests/functional/adapter/materialization/test_on_table_exists.py

@@ -80,6 +80,47 @@ def test_run_seed_test(self, project):
         check_relations_equal(project.adapter, ["seed", "materialization"])
 
 
+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 "{project.database}"."{project.test_schema}"."materialization"' in logs
+        )
+        assert "alter table" not in logs
+        results, logs = run_dbt_and_capture(["--debug", "run"], expect_pass=True)
+        assert len(results) == 1
+        assert "drop table" not in logs
+        assert "create table" 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"])
+
+
 class TestOnTableExistsRenameIncrementalFullRefresh(BaseOnTableExists):
     """
     Testing on_table_exists = `rename` configuration for incremental materialization and full refresh flag,