fix(microbatch): filter Date event_time columns without a string cast

CHANGELOG.md

@@ -5,6 +5,7 @@
 
 #### Bugs
 * `distributed_incremental` no longer creates a temporary view to read the model's structure; it is taken from the model query, which the materialization already had. Every run used to leave the view behind, and the run no longer touches that name at all, so a `<model>__dbt_view_tmp` left by an earlier version has to be dropped once by hand ([#754](https://github.com/ClickHouse/dbt-clickhouse/pull/754)).
+* Fix microbatch models failing with `Cannot convert string ... to type Date` when `event_time` is a `Date` or `Date32` column. The batch filter on upstream models compared the column with a `'YYYY-MM-DD hh:mm:ss'` string literal; it and the delete predicates on the model now use a `DateTime64` value, which also keeps batch boundaries before 1970 or after 2106 intact for `Date32` and `DateTime64` columns ([#759](https://github.com/ClickHouse/dbt-clickhouse/pull/759)).
 * Apply a materialized view model's `query_settings` to the view's own query, not only to the initial fill of the target table, so later inserts and refreshes no longer run under the server defaults. Existing views pick up the clause on the next `dbt run`, which already reissues the view query ([#755](https://github.com/ClickHouse/dbt-clickhouse/pull/755)).
 * Fix snapshots with a list `unique_key` (e.g. `unique_key=['date', 'team_id']`). The key columns were rendered as a single array literal, so every snapshot run joined all rows against all rows and inserted spurious records ([#751](https://github.com/ClickHouse/dbt-clickhouse/pull/751)).
 * Fix contract enforcement failing for `FixedString(N)` columns: the rendered data type collapsed to `String`, which no longer matches a `FixedString(N)` contract declaration ([#727](https://github.com/ClickHouse/dbt-clickhouse/pull/727)).

dbt/adapters/clickhouse/relation.py

@@ -1,4 +1,5 @@
 from dataclasses import dataclass, field
+from datetime import datetime
 from typing import Any, Dict, List, Optional, Type
 
 from dbt.adapters.base.relation import BaseRelation, EventTimeFilter, Path, Policy, Self
@@ -11,6 +12,14 @@
 NODE_TYPE_SOURCE = 'source'
 
 
+def _to_datetime_literal(value: datetime) -> str:
+    # A bare string literal is cast to the column type, and '2020-01-01 00:00:00' cannot be cast to Date
+    # or Date32. A DateTime64 value compares with Date, Date32, DateTime and DateTime64 columns alike,
+    # and unlike DateTime it is not clamped to 1970-2106, so Date32/DateTime64 columns can be batched
+    # over their whole range. Same form as the microbatch delete predicates in incremental.sql.
+    return f"toDateTime64('{value.strftime('%Y-%m-%d %H:%M:%S')}', 0)"
+
+
 @dataclass
 class ClickHouseQuotePolicy(Policy):
     database: bool = True
@@ -67,11 +76,15 @@ def _render_event_time_filtered(self, event_time_filter: EventTimeFilter) -> str
         """
         filter = ""
         if event_time_filter.start and event_time_filter.end:
-            filter = f"{event_time_filter.field_name} >= '{event_time_filter.start.strftime('%Y-%m-%d %H:%M:%S')}' and {event_time_filter.field_name} < '{event_time_filter.end.strftime('%Y-%m-%d %H:%M:%S')}'"
+            filter = f"{event_time_filter.field_name} >= {_to_datetime_literal(event_time_filter.start)} and {event_time_filter.field_name} < {_to_datetime_literal(event_time_filter.end)}"
         elif event_time_filter.start:
-            filter = f"{event_time_filter.field_name} >= '{event_time_filter.start.strftime('%Y-%m-%d %H:%M:%S')}'"
+            filter = (
+                f"{event_time_filter.field_name} >= {_to_datetime_literal(event_time_filter.start)}"
+            )
         elif event_time_filter.end:
-            filter = f"{event_time_filter.field_name} < '{event_time_filter.end.strftime('%Y-%m-%d %H:%M:%S')}'"
+            filter = (
+                f"{event_time_filter.field_name} < {_to_datetime_literal(event_time_filter.end)}"
+            )
 
         return filter
 

dbt/include/clickhouse/macros/materializations/incremental/incremental.sql

@@ -74,10 +74,10 @@
       {% do clickhouse__incremental_delete_insert(existing_relation, unique_key, incremental_predicates) %}
     {% elif incremental_strategy == 'microbatch' %}
       {%- if config.get("__dbt_internal_microbatch_event_time_start") -%}
-        {% do incremental_predicates.append(config.get("event_time") ~ " >= toDateTime('" ~ config.get("__dbt_internal_microbatch_event_time_start").strftime("%Y-%m-%d %H:%M:%S") ~ "')") %}
+        {% do incremental_predicates.append(config.get("event_time") ~ " >= toDateTime64('" ~ config.get("__dbt_internal_microbatch_event_time_start").strftime("%Y-%m-%d %H:%M:%S") ~ "', 0)") %}
       {%- endif -%}
       {%- if model.config.__dbt_internal_microbatch_event_time_end -%}
-        {% do incremental_predicates.append(config.get("event_time") ~ " < toDateTime('" ~ config.get("__dbt_internal_microbatch_event_time_end").strftime("%Y-%m-%d %H:%M:%S") ~ "')") %}
+        {% do incremental_predicates.append(config.get("event_time") ~ " < toDateTime64('" ~ config.get("__dbt_internal_microbatch_event_time_end").strftime("%Y-%m-%d %H:%M:%S") ~ "', 0)") %}
       {%- endif -%}
       {% do clickhouse__incremental_delete_insert(existing_relation, unique_key, incremental_predicates) %}
     {% elif incremental_strategy == 'append' %}

tests/integration/adapter/incremental/test_incremental_microbatch.py

@@ -1,5 +1,6 @@
 import pytest
 from dbt.tests.adapter.incremental.test_incremental_microbatch import BaseMicrobatch
+from dbt.tests.util import patch_microbatch_end_time, run_dbt
 
 _input_model_sql = """
 {{
@@ -46,3 +47,159 @@ def insert_two_rows_sql(self, project) -> str:
             database=project.database, schema=project.test_schema
         )
         return f"insert into {test_schema_relation}.input_model (id, event_time) values (4, '2020-01-04 00:00:00'), (5, '2020-01-05 00:00:00')"
+
+
+_input_model_date_sql = """
+{{
+    config(
+        materialized='table',
+        event_time='event_date'
+    )
+}}
+
+select 1 as id, toDate('2020-01-01') as event_date
+union all
+select 2 as id, toDate('2020-01-02') as event_date
+union all
+select 3 as id, toDate('2020-01-03') as event_date
+"""
+
+_microbatch_model_date_sql = """
+{{
+    config(
+        materialized='incremental',
+        incremental_strategy='microbatch',
+        unique_key='id',
+        event_time='event_date',
+        batch_size='day',
+        begin=modules.datetime.datetime(2020, 1, 1, 0, 0, 0)
+    )
+}}
+
+select * from {{ ref('input_model') }}
+"""
+
+
+class TestMicrobatchIncrementalDateEventTime(BaseMicrobatch):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "input_model.sql": _input_model_date_sql,
+            "microbatch_model.sql": _microbatch_model_date_sql,
+        }
+
+    @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_date) values (4, '2020-01-04'), (5, '2020-01-05')"
+
+
+# Date32 and DateTime64 hold dates before 1970, outside the DateTime range. The batch boundaries
+# must not be clamped to 1970-01-01, or the first batches select and delete nothing.
+_input_model_date32_sql = """
+{{
+    config(
+        materialized='table',
+        event_time='event_date'
+    )
+}}
+
+select 1 as id, toDate32('1960-01-01') as event_date
+union all
+select 2 as id, toDate32('1960-01-02') as event_date
+union all
+select 3 as id, toDate32('1960-01-03') as event_date
+"""
+
+_input_model_datetime64_sql = """
+{{
+    config(
+        materialized='table',
+        event_time='event_time'
+    )
+}}
+
+select 1 as id, toDateTime64('1960-01-01 00:00:00', 3) as event_time
+union all
+select 2 as id, toDateTime64('1960-01-02 00:00:00', 3) as event_time
+union all
+select 3 as id, toDateTime64('1960-01-03 00:00:00', 3) as event_time
+"""
+
+_microbatch_model_1960_sql = """
+{{
+    config(
+        materialized='incremental',
+        incremental_strategy='microbatch',
+        unique_key='id',
+        event_time='%s',
+        batch_size='day',
+        begin=modules.datetime.datetime(1960, 1, 1, 0, 0, 0)
+    )
+}}
+
+select * from {{ ref('input_model') }}
+"""
+
+
+class BaseMicrobatchHistorical(BaseMicrobatch):
+    """The BaseMicrobatch scenario with the clock set to January 1960."""
+
+    def test_run_with_event_time(self, project, insert_two_rows_sql):
+        with patch_microbatch_end_time("1960-01-03 13:57:00"):
+            run_dbt(["run"])
+        self.assert_row_count(project, "microbatch_model", 3)
+
+        # same day, no new data: the current batch is reprocessed without duplicating rows
+        with patch_microbatch_end_time("1960-01-03 14:57:00"):
+            run_dbt(["run"])
+        self.assert_row_count(project, "microbatch_model", 3)
+
+        project.run_sql(insert_two_rows_sql)
+        self.assert_row_count(project, "input_model", 5)
+
+        with patch_microbatch_end_time("1960-01-03 14:57:00"):
+            run_dbt(["run", "--select", "microbatch_model"])
+        self.assert_row_count(project, "microbatch_model", 3)
+
+        with patch_microbatch_end_time("1960-01-04 14:57:00"):
+            run_dbt(["run", "--select", "microbatch_model"])
+        self.assert_row_count(project, "microbatch_model", 4)
+
+        with patch_microbatch_end_time("1960-01-05 14:57:00"):
+            run_dbt(["run", "--select", "microbatch_model"])
+        self.assert_row_count(project, "microbatch_model", 5)
+
+
+class TestMicrobatchIncrementalDate32HistoricalEventTime(BaseMicrobatchHistorical):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "input_model.sql": _input_model_date32_sql,
+            "microbatch_model.sql": _microbatch_model_1960_sql % 'event_date',
+        }
+
+    @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_date) values (4, '1960-01-04'), (5, '1960-01-05')"
+
+
+class TestMicrobatchIncrementalDateTime64HistoricalEventTime(BaseMicrobatchHistorical):
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "input_model.sql": _input_model_datetime64_sql,
+            "microbatch_model.sql": _microbatch_model_1960_sql % 'event_time',
+        }
+
+    @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) values (4, '1960-01-04 00:00:00'), (5, '1960-01-05 00:00:00')"

tests/unit/test_relation.py

@@ -1,4 +1,7 @@
+from datetime import datetime
+
 import pytest
+from dbt.adapters.base.relation import EventTimeFilter
 from dbt.adapters.clickhouse.relation import ClickHouseRelation
 
 
@@ -15,3 +18,36 @@
 )
 def test_get_on_cluster(cluster, database_engine, expected):
     assert ClickHouseRelation.get_on_cluster(cluster, database_engine) is expected
+
+
+@pytest.mark.parametrize(
+    'start,end,expected',
+    [
+        (
+            datetime(2020, 1, 1),
+            datetime(2020, 1, 2),
+            "event_time >= toDateTime64('2020-01-01 00:00:00', 0)"
+            " and event_time < toDateTime64('2020-01-02 00:00:00', 0)",
+        ),
+        (
+            datetime(2020, 1, 1, 10),
+            None,
+            "event_time >= toDateTime64('2020-01-01 10:00:00', 0)",
+        ),
+        (None, datetime(2020, 1, 2), "event_time < toDateTime64('2020-01-02 00:00:00', 0)"),
+        (None, None, ""),
+        # Outside the DateTime range (1970-2106): the literal must still carry the real boundary,
+        # because toDateTime() would clamp it to 1970-01-01.
+        (
+            datetime(1960, 1, 1),
+            datetime(1960, 1, 2),
+            "event_time >= toDateTime64('1960-01-01 00:00:00', 0)"
+            " and event_time < toDateTime64('1960-01-02 00:00:00', 0)",
+        ),
+        (None, datetime(2200, 1, 1), "event_time < toDateTime64('2200-01-01 00:00:00', 0)"),
+    ],
+)
+def test_render_event_time_filtered(start, end, expected):
+    relation = ClickHouseRelation.create(schema='db', identifier='events')
+    event_time_filter = EventTimeFilter(field_name='event_time', start=start, end=end)
+    assert relation._render_event_time_filtered(event_time_filter) == expected