Repository: TobikoData/sqlmesh
Pull request #3261 (2024-10-16 20:21:22)

Fix(clickhouse): remove fractional seconds when time column is datetime/timestamp type
In some scenarios, the end of SQLMesh's incremental by time filter is created by subtracting 1 microsecond from a date, resulting in fractional seconds. For example, `2024-01-02` minus 1 microsecond = `2024-01-01 23:59:59.999999`.

Clickhouse's `DateTime` data type (and alias timestamp) only support second precision, so an error is thrown when it tries to cast fractional seconds to `DateTime`.

This PR prevents that error by removing the fractional seconds when the time column is `DateTime` or timestamp types.


Patch:
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index b2e0efc8..c29c7ce3 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -625,7 +625,15 @@ class _Model(ModelMeta, frozen=True):
 
             time_column_type = columns_to_types[self.time_column.column.name]
 
-            return to_time_column(time, time_column_type, self.time_column.format)
+            return to_time_column(
+                time,
+                time_column_type,
+                self.time_column.format,
+                include_microseconds=not (
+                    self.dialect in TIME_COLUMN_NO_MICROSECONDS_TYPES
+                    and time_column_type.is_type(*TIME_COLUMN_NO_MICROSECONDS_TYPES[self.dialect])
+                ),
+            )
         return exp.convert(time)
 
     def set_mapping_schema(self, schema: t.Dict) -> None:
@@ -2440,3 +2448,8 @@ META_FIELD_CONVERTER: t.Dict[str, t.Callable] = {
 def get_model_name(path: Path) -> str:
     path_parts = list(path.parts[path.parts.index("models") + 1 : -1]) + [path.stem]
     return ".".join(path_parts[-3:])
+
+
+TIME_COLUMN_NO_MICROSECONDS_TYPES = {
+    "clickhouse": (exp.DataType.Type.DATETIME, exp.DataType.Type.TIMESTAMP)
+}
diff --git a/sqlmesh/integrations/github/cicd/command.py b/sqlmesh/integrations/github/cicd/command.py
index 14556444..c238aea2 100644
--- a/sqlmesh/integrations/github/cicd/command.py
+++ b/sqlmesh/integrations/github/cicd/command.py
@@ -232,7 +232,7 @@ def _run_all(controller: GithubController) -> None:
                 conclusion=GithubCheckConclusion.SKIPPED,
                 skip_reason="Unit Test(s) Failed so skipping deploying to production",
             )
-        return
+        raise CICDBotError("Failed to run tests. See check status for more information.")
     pr_environment_updated = _update_pr_environment(controller)
     prod_plan_generated = False
     if pr_environment_updated:
@@ -241,8 +241,9 @@ def _run_all(controller: GithubController) -> None:
         controller.update_prod_plan_preview_check(
             status=GithubCheckStatus.COMPLETED, conclusion=GithubCheckConclusion.SKIPPED
         )
-    if tests_passed and has_required_approval and pr_environment_updated and prod_plan_generated:
-        _deploy_production(controller)
+    deployed_to_prod = False
+    if has_required_approval and prod_plan_generated:
+        deployed_to_prod = _deploy_production(controller)
     elif is_auto_deploying_prod:
         if not has_required_approval:
             skip_reason = (
@@ -263,6 +264,14 @@ def _run_all(controller: GithubController) -> None:
             conclusion=GithubCheckConclusion.SKIPPED,
             skip_reason=skip_reason,
         )
+    if (
+        not pr_environment_updated
+        or not prod_plan_generated
+        or (has_required_approval and not deployed_to_prod)
+    ):
+        raise CICDBotError(
+            "A step of the run-all check failed. See check status for more information."
+        )
 
 
 @github.command()
diff --git a/sqlmesh/utils/date.py b/sqlmesh/utils/date.py
index 419dd320..59f08802 100644
--- a/sqlmesh/utils/date.py
+++ b/sqlmesh/utils/date.py
@@ -256,14 +256,20 @@ def to_ds(obj: TimeLike) -> str:
     return to_ts(obj)[0:10]
 
 
-def to_ts(obj: TimeLike) -> str:
+def to_ts(obj: TimeLike, include_microseconds: bool = True) -> str:
     """Converts a TimeLike object into YYYY-MM-DD HH:MM:SS formatted string."""
-    return to_datetime(obj).replace(tzinfo=None).isoformat(sep=" ")
+    obj_dt = to_datetime(obj)
+    if not include_microseconds:
+        obj_dt = obj_dt.replace(microsecond=0)
+    return obj_dt.replace(tzinfo=None).isoformat(sep=" ")
 
 
-def to_tstz(obj: TimeLike) -> str:
+def to_tstz(obj: TimeLike, include_microseconds: bool = True) -> str:
     """Converts a TimeLike object into YYYY-MM-DD HH:MM:SS+00:00 formatted string."""
-    return to_datetime(obj).isoformat(sep=" ")
+    obj_dt = to_datetime(obj)
+    if not include_microseconds:
+        obj_dt = obj_dt.replace(microsecond=0)
+    return obj_dt.isoformat(sep=" ")
 
 
 def is_date(obj: TimeLike) -> bool:
@@ -289,7 +295,7 @@ def make_inclusive(start: TimeLike, end: TimeLike) -> Interval:
     In the ds ('2020-01-01') case, because start_ds and end_ds are categorical, between works even if
     start_ds and end_ds are equivalent. However, when we move to ts ('2022-01-01 12:00:00'), because timestamps
     are numeric, using simple equality doesn't make sense. When the end is not a categorical date, then it is
-    treated as an exclusive range and converted to inclusive by subtracting 1 millisecond.
+    treated as an exclusive range and converted to inclusive by subtracting 1 microsecond.
 
     Args:
         start: Start timelike object.
@@ -347,6 +353,7 @@ def to_time_column(
     time_column: t.Union[TimeLike, exp.Null],
     time_column_type: exp.DataType,
     time_column_format: t.Optional[str] = None,
+    include_microseconds: bool = True,
 ) -> exp.Expression:
     """Convert a TimeLike object to the same time format and type as the model's time column."""
     if isinstance(time_column, exp.Null):
@@ -354,9 +361,14 @@ def to_time_column(
     if time_column_type.is_type(exp.DataType.Type.DATE):
         return exp.cast(exp.Literal.string(to_ds(time_column)), to="date")
     if time_column_type.is_type(*TEMPORAL_TZ_TYPES):
-        return exp.cast(exp.Literal.string(to_tstz(time_column)), to=time_column_type)
+        return exp.cast(
+            exp.Literal.string(to_tstz(time_column, include_microseconds)),
+            to=time_column_type,
+        )
     if time_column_type.is_type(*exp.DataType.TEMPORAL_TYPES):
-        return exp.cast(exp.Literal.string(to_ts(time_column)), to=time_column_type)
+        return exp.cast(
+            exp.Literal.string(to_ts(time_column, include_microseconds)), to=time_column_type
+        )
 
     if time_column_format:
         time_column = to_datetime(time_column).strftime(time_column_format)


Test patch:
diff --git a/tests/core/engine_adapter/test_clickhouse.py b/tests/core/engine_adapter/test_clickhouse.py
index 055b70ba..12a1182f 100644
--- a/tests/core/engine_adapter/test_clickhouse.py
+++ b/tests/core/engine_adapter/test_clickhouse.py
@@ -10,6 +10,8 @@ import typing as t
 from sqlmesh.core.schema_diff import SchemaDiffer
 from datetime import datetime
 from pytest_mock.plugin import MockerFixture
+from sqlmesh.core import dialect as d
+from sqlmesh.utils.date import to_datetime
 
 pytestmark = [pytest.mark.clickhouse, pytest.mark.engine]
 
@@ -808,3 +810,49 @@ SELECT "id", "name", "price", "test_VALID_from", "test_valid_to" FROM "static" U
     """,
         dialect=adapter.dialect,
     ).sql(adapter.dialect)
+
+
+def test_to_time_column():
+    # datetime/timestamp data type should remove fractional seconds
+    expressions = d.parse(
+        """
+        MODEL (
+            name db.table,
+            kind INCREMENTAL_BY_TIME_RANGE(
+                time_column (ds)
+            ),
+            dialect clickhouse
+        );
+
+        SELECT ds::datetime
+    """
+    )
+    model = load_sql_based_model(expressions)
+    assert model.convert_to_time_column("2022-01-01 00:00:00.000001").this == d.parse_one(
+        "'2022-01-01 00:00:00'"
+    )
+    assert model.convert_to_time_column(
+        to_datetime("2022-01-01 00:00:00.000001")
+    ).this == d.parse_one("'2022-01-01 00:00:00'")
+
+    # DateTime64 data type should retain fractional seconds
+    expressions = d.parse(
+        """
+        MODEL (
+            name db.table,
+            kind INCREMENTAL_BY_TIME_RANGE(
+                time_column (ds)
+            ),
+            dialect clickhouse
+        );
+
+        SELECT ds::DateTime64
+    """
+    )
+    model = load_sql_based_model(expressions)
+    assert model.convert_to_time_column("2022-01-01 00:00:00.000001").this == d.parse_one(
+        "'2022-01-01 00:00:00.000001'"
+    )
+    assert model.convert_to_time_column(
+        to_datetime("2022-01-01 00:00:00.000001")
+    ).this == d.parse_one("'2022-01-01 00:00:00.000001'")
diff --git a/tests/integrations/github/cicd/test_github_commands.py b/tests/integrations/github/cicd/test_github_commands.py
index 222118c2..f50a76fd 100644
--- a/tests/integrations/github/cicd/test_github_commands.py
+++ b/tests/integrations/github/cicd/test_github_commands.py
@@ -16,7 +16,7 @@ from sqlmesh.integrations.github.cicd.controller import (
     GithubCheckConclusion,
     GithubCheckStatus,
 )
-from sqlmesh.utils.errors import ConflictingPlanError, PlanError, TestError
+from sqlmesh.utils.errors import ConflictingPlanError, PlanError, TestError, CICDBotError
 
 pytest_plugins = ["tests.integrations.github.cicd.fixtures"]
 pytestmark = [
@@ -461,7 +461,8 @@ def test_run_all_test_failed(
     github_output_file = tmp_path / "github_output.txt"
 
     with mock.patch.dict(os.environ, {"GITHUB_OUTPUT": str(github_output_file)}):
-        command._run_all(controller)
+        with pytest.raises(CICDBotError):
+            command._run_all(controller)
 
     assert "SQLMesh - Run Unit Tests" in controller._check_run_mapping
     test_checks_runs = controller._check_run_mapping["SQLMesh - Run Unit Tests"].all_kwargs
@@ -593,7 +594,8 @@ def test_run_all_test_exception(
     github_output_file = tmp_path / "github_output.txt"
 
     with mock.patch.dict(os.environ, {"GITHUB_OUTPUT": str(github_output_file)}):
-        command._run_all(controller)
+        with pytest.raises(CICDBotError):
+            command._run_all(controller)
 
     assert "SQLMesh - Run Unit Tests" in controller._check_run_mapping
     test_checks_runs = controller._check_run_mapping["SQLMesh - Run Unit Tests"].all_kwargs
@@ -727,7 +729,8 @@ def test_pr_update_failure(
     github_output_file = tmp_path / "github_output.txt"
 
     with mock.patch.dict(os.environ, {"GITHUB_OUTPUT": str(github_output_file)}):
-        command._run_all(controller)
+        with pytest.raises(CICDBotError):
+            command._run_all(controller)
 
     assert "SQLMesh - PR Environment Synced" in controller._check_run_mapping
     pr_checks_runs = controller._check_run_mapping["SQLMesh - PR Environment Synced"].all_kwargs
@@ -851,7 +854,8 @@ def make_test_prod_update_failure_case(
     github_output_file = tmp_path / "github_output.txt"
 
     with mock.patch.dict(os.environ, {"GITHUB_OUTPUT": str(github_output_file)}):
-        command._run_all(controller)
+        with pytest.raises(CICDBotError):
+            command._run_all(controller)
 
     assert "SQLMesh - Prod Plan Preview" in controller._check_run_mapping
     prod_plan_preview_checks_runs = controller._check_run_mapping[
diff --git a/tests/integrations/github/cicd/test_integration.py b/tests/integrations/github/cicd/test_integration.py
index 51a0367b..1c8b4aaa 100644
--- a/tests/integrations/github/cicd/test_integration.py
+++ b/tests/integrations/github/cicd/test_integration.py
@@ -23,6 +23,7 @@ from sqlmesh.integrations.github.cicd.controller import (
     GithubCheckStatus,
     GithubController,
 )
+from sqlmesh.utils.errors import CICDBotError
 from tests.integrations.github.cicd.fixtures import MockIssueComment
 
 pytest_plugins = ["tests.integrations.github.cicd.fixtures"]
@@ -509,7 +510,8 @@ def test_merge_pr_has_non_breaking_change_no_categorization(
     github_output_file = tmp_path / "github_output.txt"
 
     with mock.patch.dict(os.environ, {"GITHUB_OUTPUT": str(github_output_file)}):
-        command._run_all(controller)
+        with pytest.raises(CICDBotError):
+            command._run_all(controller)
 
     assert "SQLMesh - Run Unit Tests" in controller._check_run_mapping
     test_checks_runs = controller._check_run_mapping["SQLMesh - Run Unit Tests"].all_kwargs
@@ -1347,7 +1349,8 @@ def test_error_msg_when_applying_plan_with_bug(
     github_output_file = tmp_path / "github_output.txt"
 
     with mock.patch.dict(os.environ, {"GITHUB_OUTPUT": str(github_output_file)}):
-        command._run_all(controller)
+        with pytest.raises(CICDBotError):
+            command._run_all(controller)
 
     assert "SQLMesh - Run Unit Tests" in controller._check_run_mapping
     test_checks_runs = controller._check_run_mapping["SQLMesh - Run Unit Tests"].all_kwargs