Use timestamp_tz type in microbatch delete DDL (#1257)

Author: MichelleArk

.changes/unreleased/Fixes-20241127-162204.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Use timestamp_tz type in microbatch `delete` DDL
+time: 2024-11-27T16:22:04.103212-05:00
+custom:
+  Author: michelleark
+  Issue: "1256"

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

@@ -58,10 +58,10 @@
 
     {#-- Add additional incremental_predicates to filter for batch --#}
     {% if model.config.get("__dbt_internal_microbatch_event_time_start") -%}
-        {% do incremental_predicates.append("DBT_INTERNAL_TARGET." ~ model.config.event_time ~ " >= TIMESTAMP '" ~ model.config.__dbt_internal_microbatch_event_time_start ~ "'") %}
+        {% do incremental_predicates.append("DBT_INTERNAL_TARGET." ~ model.config.event_time ~ " >= to_timestamp_tz('" ~ model.config.__dbt_internal_microbatch_event_time_start ~ "')") %}
     {% endif %}
     {% if model.config.__dbt_internal_microbatch_event_time_end -%}
-        {% do incremental_predicates.append("DBT_INTERNAL_TARGET." ~ model.config.event_time ~ " < TIMESTAMP '" ~ model.config.__dbt_internal_microbatch_event_time_end ~ "'") %}
+        {% do incremental_predicates.append("DBT_INTERNAL_TARGET." ~ model.config.event_time ~ " < to_timestamp_tz('" ~ model.config.__dbt_internal_microbatch_event_time_end ~ "')") %}
     {% endif %}
     {% do arg_dict.update({'incremental_predicates': incremental_predicates}) %}
 

tests/functional/adapter/test_incremental_microbatch.py

@@ -3,6 +3,16 @@
     BaseMicrobatch,
 )
 
+# Create input with UTC timestamps
+_input_model_sql = """
+{{ config(materialized='table', event_time='event_time') }}
+select 1 as id, to_timestamp_tz('2020-01-01 00:00:00-0') as event_time
+union all
+select 2 as id, to_timestamp_tz('2020-01-02 00:00:00-0') as event_time
+union all
+select 3 as id, to_timestamp_tz('2020-01-03 00:00:00-0') as event_time
+"""
+
 
 # No requirement for a unique_id for snowflake microbatch!
 _microbatch_model_no_unique_id_sql = """
@@ -16,6 +26,10 @@ class TestSnowflakeMicrobatch(BaseMicrobatch):
     def microbatch_model_sql(self) -> str:
         return _microbatch_model_no_unique_id_sql
 
+    @pytest.fixture(scope="class")
+    def input_model_sql(self) -> str:
+        return _input_model_sql
+
     @pytest.fixture(scope="class")
     def insert_two_rows_sql(self, project) -> str:
         test_schema_relation = project.adapter.Relation.create(