Repository: TobikoData/sqlmesh
Pull request #3582 (2025-01-03 22:59:23)

Fix(clickhouse): specify DateTime64 precision for model partitioning expression
SQLMesh automatically partitions Clickhouse incremental by time range models, aggregating timestamps to week. If the timestamp data type is not a datetime variant, SQLMesh automatically casts it to DateTime64 in the aggregation expression.

In Clickhouse, casting to DateTime64 type requires specifying a numeric precision.

This PR specifies a precision of 9/nanoseconds (using max precision to prevent accidental truncation of timestamp digits). 

swt30: :+1: only caveat I can see is that the [maximum date representable in a DateTime64(9)](https://clickhouse.com/docs/en/sql-reference/data-types/datetime64) is `2262-04-11 23:47:16`. But I won't be around to see the Y2K bug caused by that :laughing: so using the maximum supported precision feels like the way to go. 

This only applies when you pass a non-conformable type like a string or an integer anyway, right? 
treysp: > This only applies when you pass a non-conformable type like a string or an integer anyway, right?

Yep, if time column is non-conformable AND you don't include the time column in the model partitioning expression

Patch:
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index 6f8bf10e..f53e1ffb 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -2290,7 +2290,7 @@ def clickhouse_partition_func(
     if col_type.is_type(exp.DataType.Type.UNKNOWN):
         return exp.func(
             "toMonday",
-            exp.cast(column, exp.DataType.build("DateTime64('UTC')", dialect="clickhouse")),
+            exp.cast(column, exp.DataType.build("DateTime64(9, 'UTC')", dialect="clickhouse")),
             dialect="clickhouse",
         )
 
@@ -2298,7 +2298,7 @@ def clickhouse_partition_func(
     return exp.cast(
         exp.func(
             "toMonday",
-            exp.cast(column, exp.DataType.build("DateTime64('UTC')", dialect="clickhouse")),
+            exp.cast(column, exp.DataType.build("DateTime64(9, 'UTC')", dialect="clickhouse")),
             dialect="clickhouse",
         ),
         col_type,
diff --git a/sqlmesh/core/plan/evaluator.py b/sqlmesh/core/plan/evaluator.py
index 8a39c54c..8893cb77 100644
--- a/sqlmesh/core/plan/evaluator.py
+++ b/sqlmesh/core/plan/evaluator.py
@@ -223,12 +223,6 @@ class BuiltInPlanEvaluator(PlanEvaluator):
             for s in snapshots.values()
             if s.is_model and not s.is_symbolic and plan.is_selected_for_backfill(s.name)
         ]
-        snapshots_to_create_count = len(snapshots_to_create)
-
-        if snapshots_to_create_count > 0:
-            self.console.start_creation_progress(
-                snapshots_to_create_count, plan.environment, self.default_catalog
-            )
 
         completed = False
         try:
@@ -237,6 +231,9 @@ class BuiltInPlanEvaluator(PlanEvaluator):
                 snapshots,
                 allow_destructive_snapshots=plan.allow_destructive_models,
                 deployability_index=deployability_index,
+                on_start=lambda x: self.console.start_creation_progress(
+                    x, plan.environment, self.default_catalog
+                ),
                 on_complete=self.console.update_creation_progress,
             )
             completed = True
diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py
index c15bc661..e4e00ef6 100644
--- a/sqlmesh/core/snapshot/evaluator.py
+++ b/sqlmesh/core/snapshot/evaluator.py
@@ -261,6 +261,7 @@ class SnapshotEvaluator:
         target_snapshots: t.Iterable[Snapshot],
         snapshots: t.Dict[SnapshotId, Snapshot],
         deployability_index: t.Optional[DeployabilityIndex] = None,
+        on_start: t.Optional[t.Callable] = None,
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
         allow_destructive_snapshots: t.Set[str] = set(),
     ) -> None:
@@ -270,6 +271,7 @@ class SnapshotEvaluator:
             target_snapshots: Target snapshots.
             snapshots: Mapping of snapshot ID to snapshot.
             deployability_index: Determines snapshots that are deployable in the context of this creation.
+            on_start: A callback to initialize the snapshot creation progress bar.
             on_complete: A callback to call on each successfully created snapshot.
             allow_destructive_snapshots: Set of snapshots that are allowed to have destructive schema changes.
         """
@@ -326,11 +328,11 @@ class SnapshotEvaluator:
                         table_deployability[table_name]
                     )
                 target_deployability_flags[snapshot.name].sort()
-            elif on_complete:
-                on_complete(snapshot)
 
         if not snapshots_to_create:
             return
+        if on_start:
+            on_start(len(snapshots_to_create))
         self._create_schemas(tables_by_schema, gateway_by_schema)
         self._create_snapshots(
             snapshots_to_create,
@@ -350,7 +352,7 @@ class SnapshotEvaluator:
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]],
         allow_destructive_snapshots: t.Set[str],
     ) -> None:
-        """Internal method to create tables in parrallel."""
+        """Internal method to create tables in parallel."""
         with self.concurrent_context():
             concurrent_apply_to_snapshots(
                 snapshots_to_create,


Test patch:
diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py
index f1dc54b4..7d0eea69 100644
--- a/tests/cli/test_cli.py
+++ b/tests/cli/test_cli.py
@@ -220,7 +220,8 @@ def test_plan_restate_model(runner, tmp_path):
     assert_duckdb_test(result)
     assert "No changes to plan: project files match the `prod` environment" in result.output
     assert "sqlmesh_example.full_model evaluated in" in result.output
-    assert_backfill_success(result)
+    assert_model_batches_executed(result)
+    assert_target_env_updated(result)
 
 
 @pytest.mark.parametrize("flag", ["--skip-backfill", "--dry-run"])
@@ -348,7 +349,6 @@ def test_plan_dev_create_from_virtual(runner, tmp_path):
     )
     assert result.exit_code == 0
     assert_new_env(result, "dev2", "dev", initialize=False)
-    assert_model_versions_created(result)
     assert_target_env_updated(result)
     assert_virtual_update(result)
 
diff --git a/tests/core/engine_adapter/integration/test_integration_clickhouse.py b/tests/core/engine_adapter/integration/test_integration_clickhouse.py
index 15d7da03..4a97eb7d 100644
--- a/tests/core/engine_adapter/integration/test_integration_clickhouse.py
+++ b/tests/core/engine_adapter/integration/test_integration_clickhouse.py
@@ -497,3 +497,51 @@ def test_insert_overwrite_by_condition_inc_by_partition(ctx: TestContext):
             ]
         ),
     )
+
+
+def test_inc_by_time_auto_partition_string(ctx: TestContext):
+    # ensure automatic time partitioning works when the time column is not a Date/DateTime type
+    existing_table_name = _create_table_and_insert_existing_data(
+        ctx,
+        columns_to_types={
+            "id": exp.DataType.build("Int8", "clickhouse"),
+            "ds": exp.DataType.build("String", "clickhouse"),  # String time column
+        },
+        table_name="data_existing",
+        partitioned_by=None,
+    )
+
+    sqlmesh_context, model = ctx.upsert_sql_model(
+        f"""
+        MODEL (
+            name test.inc_by_time_no_partition,
+            kind INCREMENTAL_BY_TIME_RANGE (
+                time_column ds
+            ),
+            dialect clickhouse,
+            start '2023-01-01'
+        );
+
+        SELECT
+            id::Int8,
+            ds::String
+        FROM {existing_table_name.sql()}
+        WHERE ds BETWEEN @start_ds AND @end_ds
+        """
+    )
+
+    plan = sqlmesh_context.plan(no_prompts=True, auto_apply=True)
+
+    physical_location = ctx.engine_adapter.get_data_objects(
+        plan.environment.snapshots[0].physical_schema
+    )[0]
+
+    partitions = ctx.engine_adapter.fetchall(
+        exp.select("_partition_id")
+        .distinct()
+        .from_(f"{physical_location.schema_name}.{physical_location.name}")
+    )
+
+    # The automatic time partitioning creates one partition per week. The 4 input data points
+    # are located in three distinct weeks, which should have one partition each.
+    assert len(partitions) == 3
diff --git a/tests/core/engine_adapter/test_clickhouse.py b/tests/core/engine_adapter/test_clickhouse.py
index 19f4df38..ec4c0992 100644
--- a/tests/core/engine_adapter/test_clickhouse.py
+++ b/tests/core/engine_adapter/test_clickhouse.py
@@ -371,7 +371,8 @@ def test_partitioned_by_expr(make_mocked_engine_adapter: t.Callable):
     )
 
     assert (
-        model.partitioned_by[0].sql("clickhouse") == """toMonday(CAST("ds" AS DateTime64('UTC')))"""
+        model.partitioned_by[0].sql("clickhouse")
+        == """toMonday(CAST("ds" AS DateTime64(9, 'UTC')))"""
     )
 
     # user specifies without time column, unknown time column type
@@ -396,7 +397,7 @@ def test_partitioned_by_expr(make_mocked_engine_adapter: t.Callable):
     )
 
     assert [p.sql("clickhouse") for p in model.partitioned_by] == [
-        """toMonday(CAST("ds" AS DateTime64('UTC')))""",
+        """toMonday(CAST("ds" AS DateTime64(9, 'UTC')))""",
         '"x"',
     ]
 
@@ -444,7 +445,7 @@ def test_partitioned_by_expr(make_mocked_engine_adapter: t.Callable):
 
     assert (
         model.partitioned_by[0].sql("clickhouse")
-        == """CAST(toMonday(CAST("ds" AS DateTime64('UTC'))) AS String)"""
+        == """CAST(toMonday(CAST("ds" AS DateTime64(9, 'UTC'))) AS String)"""
     )
 
     # user specifies partitioned_by with time column