Feat: add dtntz date macro variant

docs/concepts/macros/macro_variables.md

@@ -68,6 +68,7 @@ Prefixes:
 Postfixes:
 
 * dt - A python datetime object that converts into a native SQL `TIMESTAMP` (or SQL engine equivalent)
+* dtntz - A python datetime object that converts into a native SQL `TIMESTAMP WITHOUT TIME ZONE` (or SQL engine equivalent)
 * date - A python date object that converts into a native SQL `DATE`
 * ds - A date string with the format: '%Y-%m-%d'
 * ts - An ISO 8601 datetime formatted string: '%Y-%m-%d %H:%M:%S'
@@ -83,6 +84,11 @@ All predefined temporal macro variables:
     * @end_dt
     * @execution_dt
 
+* dtntz
+    * @start_dtntz
+    * @end_dtntz
+    * @execution_dtntz
+
 * date
     * @start_date
     * @end_date

docs/integrations/engines/redshift.md

@@ -33,3 +33,37 @@ pip install "sqlmesh[redshift]"
 | `serverless_acct_id`    | The account ID of the serverless cluster                                                                    | string |    N     |
 | `serverless_work_group` | The name of work group for serverless end point                                                             | string |    N     |
 | `enable_merge`         | Whether the incremental_by_unique_key model kind will use the native Redshift MERGE operation or SQLMesh's logical merge. (Default: `False`)           |  bool  |    N     |
+
+## Performance Considerations
+
+### Timestamp Macro Variables and Sort Keys
+
+When working with Redshift tables that have a `TIMESTAMP` sort key, using the standard `@start_dt` and `@end_dt` macro variables may lead to performance issues. These macros render as `TIMESTAMP WITH TIME ZONE` values in SQL queries, which prevents Redshift from performing efficient pruning when filtering against `TIMESTAMP` (without timezone) sort keys.
+
+This can result in full table scans instead, causing significant performance degradation.
+
+**Solution**: Use the `_dtntz` (datetime no timezone) variants of macro variables:
+
+- `@start_dtntz` instead of `@start_dt`
+- `@end_dtntz` instead of `@end_dt`
+
+These variants render as `TIMESTAMP WITHOUT TIME ZONE`, allowing Redshift to properly utilize sort key optimizations.
+
+**Example**:
+
+```sql linenums="1"
+-- Inefficient: May cause full table scan
+SELECT * FROM my_table
+WHERE timestamp_column >= @start_dt
+  AND timestamp_column < @end_dt
+
+-- Efficient: Uses sort key optimization
+SELECT * FROM my_table
+WHERE timestamp_column >= @start_dtntz
+  AND timestamp_column < @end_dtntz
+
+-- Alternative: Cast to timestamp
+SELECT * FROM my_table
+WHERE timestamp_column >= @start_ts::timestamp
+  AND timestamp_column < @end_ts::timestamp
+```

sqlmesh/utils/date.py

@@ -253,15 +253,20 @@ def date_dict(
 
     for prefix, time_like in prefixes:
         dt = to_datetime(time_like)
+        dtntz = dt.replace(tzinfo=None)
+
         millis = to_timestamp(time_like)
+
         kwargs[f"{prefix}_dt"] = dt
+        kwargs[f"{prefix}_dtntz"] = dtntz
         kwargs[f"{prefix}_date"] = to_date(dt)
         kwargs[f"{prefix}_ds"] = to_ds(time_like)
         kwargs[f"{prefix}_ts"] = to_ts(dt)
         kwargs[f"{prefix}_tstz"] = to_tstz(dt)
         kwargs[f"{prefix}_epoch"] = millis / 1000
         kwargs[f"{prefix}_millis"] = millis
         kwargs[f"{prefix}_hour"] = dt.hour
+
     return kwargs
 
 

tests/core/test_model.py

@@ -10518,3 +10518,28 @@ def test_boolean_property_validation() -> None:
     )
     model = load_sql_based_model(expressions, dialect="tsql")
     assert model.enabled
+
+
+def test_datetime_without_timezone_variable_redshift() -> None:
+    expressions = d.parse(
+        """
+        MODEL (
+            name test,
+            kind INCREMENTAL_BY_TIME_RANGE (
+                time_column test_time_col,
+                batch_size 1,
+                batch_concurrency 1
+            ),
+            start '2025-06-01',
+            dialect redshift
+        );
+
+        SELECT @start_dtntz AS test_time_col
+        """
+    )
+    model = load_sql_based_model(expressions, dialect="redshift")
+
+    assert (
+        model.render_query_or_raise().sql("redshift")
+        == '''SELECT CAST('1970-01-01 00:00:00' AS TIMESTAMP) AS "test_time_col"'''
+    )

tests/utils/test_date.py

@@ -258,6 +258,10 @@ def test_date_dict():
         "execution_dt": datetime(2020, 1, 2, 1, 0, 0, tzinfo=UTC),
         "start_dt": datetime(2020, 1, 1, 0, 0, 0, tzinfo=UTC),
         "end_dt": datetime(2020, 1, 2, 0, 0, 0, tzinfo=UTC),
+        "latest_dtntz": datetime(2020, 1, 2, 1, 0, 0, tzinfo=None),
+        "execution_dtntz": datetime(2020, 1, 2, 1, 0, 0, tzinfo=None),
+        "start_dtntz": datetime(2020, 1, 1, 0, 0, 0, tzinfo=None),
+        "end_dtntz": datetime(2020, 1, 2, 0, 0, 0, tzinfo=None),
         "latest_date": date(2020, 1, 2),
         "execution_date": date(2020, 1, 2),
         "start_date": date(2020, 1, 1),