fix(dlt): use timestamp macros for incremental model time filter (#5867)

Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com>

Author: anxkhn

sqlmesh/integrations/dlt.py

@@ -208,7 +208,7 @@ def generate_incremental_model(
 FROM
   {from_clause}
 WHERE
-  {time_column} BETWEEN @start_ds AND @end_ds
+  {time_column} BETWEEN @start_ts AND @end_ts
 """
 
 

tests/cli/test_cli.py

@@ -921,7 +921,7 @@ def test_dlt_filesystem_pipeline(tmp_path):
 FROM
   filesystem_pipeline_dataset.equipment as c
 WHERE
-  TO_TIMESTAMP(CAST(c._dlt_load_id AS DOUBLE)) BETWEEN @start_ds AND @end_ds
+  TO_TIMESTAMP(CAST(c._dlt_load_id AS DOUBLE)) BETWEEN @start_ts AND @end_ts
 """
 
     with open(equipment_model_path) as file:
@@ -1064,7 +1064,7 @@ def test_dlt_pipeline(runner, tmp_path):
 FROM
   sushi_dataset.sushi_types as c
 WHERE
-  TO_TIMESTAMP(CAST(c._dlt_load_id AS DOUBLE)) BETWEEN @start_ds AND @end_ds
+  TO_TIMESTAMP(CAST(c._dlt_load_id AS DOUBLE)) BETWEEN @start_ts AND @end_ts
 """
 
     dlt_sushi_types_model_path = tmp_path / "models/incremental_sushi_types.sql"
@@ -1095,7 +1095,7 @@ def test_dlt_pipeline(runner, tmp_path):
 FROM
   sushi_dataset._dlt_loads as c
 WHERE
-  TO_TIMESTAMP(CAST(c.load_id AS DOUBLE)) BETWEEN @start_ds AND @end_ds
+  TO_TIMESTAMP(CAST(c.load_id AS DOUBLE)) BETWEEN @start_ts AND @end_ts
 """
 
     with open(dlt_loads_model_path) as file:
@@ -1122,7 +1122,7 @@ def test_dlt_pipeline(runner, tmp_path):
 ON
   c._dlt_parent_id = p._dlt_id
 WHERE
-  TO_TIMESTAMP(CAST(p._dlt_load_id AS DOUBLE)) BETWEEN @start_ds AND @end_ds
+  TO_TIMESTAMP(CAST(p._dlt_load_id AS DOUBLE)) BETWEEN @start_ts AND @end_ts
 """
 
     with open(dlt_sushi_fillings_model_path) as file:

tests/integrations/test_dlt.py

@@ -0,0 +1,21 @@
+from sqlmesh.integrations.dlt import generate_incremental_model
+
+
+def test_generate_incremental_model_filters_on_timestamp_macros() -> None:
+    # The DLT-generated model's time column is a timestamp
+    # (TO_TIMESTAMP(...)). It must therefore be filtered with the inclusive
+    # timestamp macros @start_ts/@end_ts, not the categorical date macros
+    # @start_ds/@end_ds, which both render midnight and exclude any rows past
+    # 00:00:00 on a single-day run.
+    model = generate_incremental_model(
+        "dataset_sqlmesh.incremental_equipment",
+        "  CAST(c.item_id AS BIGINT) AS item_id",
+        "",
+        "dataset.equipment",
+        "duckdb",
+        "c._dlt_load_id",
+    )
+
+    assert "BETWEEN @start_ts AND @end_ts" in model
+    assert "@start_ds" not in model
+    assert "@end_ds" not in model