Add an update to allow microbatching to handle the datetime data type. This is not supported in Fabric so isn't covered by the upstream adapater.

DBT passes in event times using a UTC string that datetime column types cannot parse.

By converting this to a datetimeoffset it can be filtered against datetime, datetime2 and datetimeoffset columns which can exist in SQL Server.

Author: Benjamin-Knight

dbt/adapters/sqlserver/sqlserver_relation.py

@@ -1,7 +1,7 @@
 from dataclasses import dataclass, field
 from typing import Optional, Type
 
-from dbt.adapters.base.relation import BaseRelation
+from dbt.adapters.base.relation import BaseRelation, EventTimeFilter
 from dbt.adapters.utils import classproperty
 from dbt_common.exceptions import DbtRuntimeError
 
@@ -49,3 +49,28 @@ def __post_init__(self):
 
     def relation_max_name_length(self):
         return MAX_CHARACTERS_IN_IDENTIFIER
+
+    def _render_event_time_filtered(self, event_time_filter: EventTimeFilter) -> str:
+        """
+        Returns "" if start and end are both None
+        """
+        filter = ""
+        if event_time_filter.start and event_time_filter.end:
+            filter = (
+                f"{event_time_filter.field_name} >="
+                f" parse('{event_time_filter.start}' as datetimeoffset)"
+                f" and {event_time_filter.field_name} <"
+                f" parse('{event_time_filter.end}' as datetimeoffset)"
+            )
+        elif event_time_filter.start:
+            filter = (
+                f"{event_time_filter.field_name} >="
+                f" parse('{event_time_filter.start}' as datetimeoffset)"
+            )
+        elif event_time_filter.end:
+            filter = (
+                f"{event_time_filter.field_name} <"
+                f" parse('{event_time_filter.end}' as datetimeoffset)"
+            )
+
+        return filter

dbt/include/sqlserver/macros/materializations/models/incremental/merge.sql

@@ -0,0 +1,31 @@
+{% macro sqlserver__get_incremental_microbatch_sql(arg_dict) %}
+    {%- set target = arg_dict["target_relation"] -%}
+    {%- set source = arg_dict["temp_relation"] -%}
+    {%- set dest_columns = arg_dict["dest_columns"] -%}
+    {%- set incremental_predicates = [] if arg_dict.get('incremental_predicates') is none else arg_dict.get('incremental_predicates') -%}
+
+    {#-- Add additional incremental_predicates to filter for batch --#}
+    {% if model.config.get("__dbt_internal_microbatch_event_time_start") -%}
+    {{ log("incremenal append event start time > DBT_INTERNAL_TARGET." ~ model.config.event_time ~ " >= parse('" ~ model.config.__dbt_internal_microbatch_event_time_start ~ "' as datetimeoffset)") }}
+        {% do incremental_predicates.append("DBT_INTERNAL_TARGET." ~ model.config.event_time ~ " >= parse('" ~ model.config.__dbt_internal_microbatch_event_time_start ~ "' as datetimeoffset)") %}
+    {% endif %}
+    {% if model.config.__dbt_internal_microbatch_event_time_end -%}
+    {{ log("incremenal append event end time < DBT_INTERNAL_TARGET." ~ model.config.event_time ~ " < parse('" ~ model.config.__dbt_internal_microbatch_event_time_end ~ "' as datetimeoffset)") }}
+        {% do incremental_predicates.append("DBT_INTERNAL_TARGET." ~ model.config.event_time ~ " < parse('" ~ model.config.__dbt_internal_microbatch_event_time_end ~ "' as datetimeoffset)") %}
+    {% endif %}
+    {% do arg_dict.update({'incremental_predicates': incremental_predicates}) %}
+
+    delete DBT_INTERNAL_TARGET from {{ target }} AS DBT_INTERNAL_TARGET
+    where (
+    {% for predicate in incremental_predicates %}
+        {%- if not loop.first %}and {% endif -%} {{ predicate }}
+    {% endfor %}
+    );
+
+    {%- set dest_cols_csv = get_quoted_csv(dest_columns | map(attribute="name")) -%}
+    insert into {{ target }} ({{ dest_cols_csv }})
+    (
+        select {{ dest_cols_csv }}
+        from {{ source }}
+    )
+{% endmacro %}

tests/functional/adapter/dbt/test_incremental_microbatch_datetime.py

@@ -0,0 +1,47 @@
+import pytest
+from dbt.tests.adapter.incremental.test_incremental_microbatch import BaseMicrobatch
+
+_microbatch_model_no_unique_id_sql_datetime = """
+{{ config(materialized='incremental', incremental_strategy='microbatch',
+event_time='event_time', batch_size='day', begin='2020-01-01 00:00:00') }}
+select * from {{ ref('input_model') }}
+"""
+
+_input_model_sql_datetime = """
+{{ config(materialized='table', event_time='event_time') }}
+select 1 as id, '2020-01-01 00:00:00' as event_time
+union all
+select 2 as id, '2020-01-02 00:00:00' as event_time
+union all
+select 3 as id, '2020-01-03 00:00:00' as event_time
+"""
+
+
+class TestSQLServerMicrobatchDateTime(BaseMicrobatch):
+    """
+    Setup a version of the microbatch testing that uses a datetime column as the event_time
+    This is to test that the microbatch strategy can handle datetime columns when passing in
+    event times as UTC strings
+    """
+
+    @pytest.fixture(scope="class")
+    def microbatch_model_sql(self) -> str:
+        return _microbatch_model_no_unique_id_sql_datetime
+
+    @pytest.fixture(scope="class")
+    def input_model_sql(self) -> str:
+        """
+        This is the SQL that defines the input model to the microbatch model,
+        including any {{ config(..) }}. event_time is a required configuration of this input
+        """
+        return _input_model_sql_datetime
+
+    @pytest.fixture(scope="class")
+    def insert_two_rows_sql(self, project) -> str:
+        test_schema_relation = project.adapter.Relation.create(
+            database=project.database, schema=project.test_schema
+        )
+        return (
+            f"insert into {test_schema_relation}.input_model (id, event_time) "
+            f"values (4, '2020-01-04 00:00:00'), (5, '2020-01-05 00:00:00')"
+        )