Repository: TobikoData/sqlmesh
Pull request #3138 (2024-09-16 16:50:56)

feat: Add SQLMesh warning when an exp.Command is parsed
[Public slack context](https://tobiko-data.slack.com/archives/C044BRE5W4S/p1726478874076549)


This PR overrides SQLGlot's `_warn_unsupported()` to suggest the usage of Jinja on the occasion that the (unsupported) statement is attempting to reference the physical table.




Patch:
diff --git a/docs/concepts/macros/macro_variables.md b/docs/concepts/macros/macro_variables.md
index c23310b4..4a7b697f 100644
--- a/docs/concepts/macros/macro_variables.md
+++ b/docs/concepts/macros/macro_variables.md
@@ -53,7 +53,11 @@ Most predefined variables are related to time and use a combination of prefixes
 
 SQLMesh uses the python [datetime module](https://docs.python.org/3/library/datetime.html) for handling dates and times. It uses the standard [Unix epoch](https://en.wikipedia.org/wiki/Unix_time) start of 1970-01-01.
 
-*All predefined variables with a time component use the [UTC time zone](https://en.wikipedia.org/wiki/Coordinated_Universal_Time).*
+!!! tip "Important"
+
+    Predefined variables with a time component always use the [UTC time zone](https://en.wikipedia.org/wiki/Coordinated_Universal_Time).
+
+    Learn more about timezones and incremental models [here](../models/model_kinds.md#timezones).
 
 Prefixes:
 
diff --git a/docs/concepts/models/model_kinds.md b/docs/concepts/models/model_kinds.md
index 198daf92..8b015687 100644
--- a/docs/concepts/models/model_kinds.md
+++ b/docs/concepts/models/model_kinds.md
@@ -21,8 +21,20 @@ MODEL (
 );
 ```
 
+<a id="timezones"></a>
 In addition to specifying a time column in the `MODEL` DDL, the model's query must contain a `WHERE` clause that filters the upstream records by time range. SQLMesh provides special macros that represent the start and end of the time range being processed: `@start_date` / `@end_date` and `@start_ds` / `@end_ds`. Refer to [Macros](../macros/macro_variables.md) for more information.
 
+!!! tip "Important"
+
+    A model's `time_column` should be in the [UTC time zone](https://en.wikipedia.org/wiki/Coordinated_Universal_Time) to ensure correct interaction with SQLMesh's scheduler and predefined macro variables.
+
+    This requirement aligns with the data engineering best practice of converting datetime/timestamp columns to UTC as soon as they are ingested into the data system and only converting them to local timezones when they exit the system for downstream uses.
+
+    Placing all timezone conversion code in the system's first/last transformation models prevents inadvertent timezone-related errors as data flows between models.
+
+    If a model must use a different timezone, parameters like [lookback](./overview.md#lookback), [allow_partials](./overview.md#allow_partials), and [cron](./overview.md#cron) with offset time can be used to try to account for misalignment between the model's timezone and the UTC timezone used by SQLMesh.
+
+
 This example implements a complete `INCREMENTAL_BY_TIME_RANGE` model that specifies the time column name `event_date` in the `MODEL` DDL and includes a SQL `WHERE` clause to filter records by time range:
 
 ```sql linenums="1" hl_lines="3-5 12-13"
@@ -44,6 +56,10 @@ WHERE
 ### Time column
 SQLMesh needs to know which column in the model's output represents the timestamp or date associated with each record.
 
+!!! tip "Important"
+
+    The `time_column` variable should be in the UTC time zone - learn more [above](#timezones).
+
 The time column is used to determine which records will be overwritten during data [restatement](../plans.md#restatement-plans) and provides a partition key for engines that support partitioning (such as Apache Spark). The name of the time column is specified in the `MODEL` DDL `kind` specification:
 
 ```sql linenums="1" hl_lines="4"
@@ -64,7 +80,10 @@ MODEL (
   )
 );
 ```
-**Note:** The time format should be defined using the same SQL dialect as the one used to define the model's query.
+
+!!! note
+
+    The time format should be defined using the same SQL dialect as the one used to define the model's query.
 
 SQLMesh also uses the time column to automatically append a time range filter to the model's query at runtime, which prevents records that are not part of the target interval from being stored. This is a safety mechanism that prevents unintentionally overwriting unrelated records when handling late-arriving data.
 
@@ -217,7 +236,7 @@ MODEL (
 );
 ```
 
-`INCREMENTAL_BY_UNIQUE_KEY` model kinds can also filter upstream records by time range using a SQL `WHERE` clause and the `@start_date`, `@end_date` or other macros (similar to the [INCREMENTAL_BY_TIME_RANGE](#incremental_by_time_range) kind):
+`INCREMENTAL_BY_UNIQUE_KEY` model kinds can also filter upstream records by time range using a SQL `WHERE` clause and the `@start_date`, `@end_date` or other macro variables (similar to the [INCREMENTAL_BY_TIME_RANGE](#incremental_by_time_range) kind). Note that SQLMesh macro time variables are in the UTC time zone.
 ```sql linenums="1" hl_lines="6-7"
 SELECT
   name::TEXT as name,
diff --git a/docs/concepts/models/overview.md b/docs/concepts/models/overview.md
index 5f7adfac..52a2281b 100644
--- a/docs/concepts/models/overview.md
+++ b/docs/concepts/models/overview.md
@@ -348,6 +348,10 @@ Some properties are only available in specific model kinds - see the [model conf
 
     Engines that support partitioning, such as Spark and BigQuery, use the time column as the model's partition key. Multi-column partitions or modifications to columns can be specified with the [`partitioned_by` property](#partitioned_by).
 
+    !!! tip "Important"
+
+        The `time_column` variable should be in the UTC time zone - learn more [here](./model_kinds.md#timezones).
+
 ### batch_size
 :   Batch size is used to optimize backfilling incremental data. It determines the maximum number of intervals to run in a single job.
 
@@ -410,36 +414,3 @@ FROM y;
 -- Cleanup statements
 DROP TABLE temp_table;
 ```
-
-## Time column
-Models that are loaded incrementally require a time column to partition data.
-
-A time column is a column in a model with an optional format string in the dialect of the model; for example, `'%Y-%m-%d'` for DuckDB or `'yyyy-mm-dd'` for Snowflake.
-
-For more information, refer to [time column](./model_kinds.md#time-column).
-
-### Advanced usage
-The column used as your model's time column is not limited to a text or date type. In the following example, the time column, `di`, is an integer:
-
-```sql linenums="1" hl_lines="5"
--- Orders are partitioned by the di int column
-MODEL (
-  name sushi.orders,
-  dialect duckdb,
-  kind INCREMENTAL_BY_TIME_RANGE (
-    time_column (order_date_int, '%Y%m%d')
-  ),
-);
-
-SELECT
-  id::INT AS id, -- Primary key
-  customer_id::INT AS customer_id, -- Id of customer who made the order
-  waiter_id::INT AS waiter_id, -- Id of waiter who took the order
-  start_ts::TEXT AS start_ts, -- Start timestamp
-  end_ts::TEXT AS end_ts, -- End timestamp
-  di::INT AS order_date_int -- Date of order
-FROM raw.orders
-WHERE
-  order_date_int BETWEEN @start_ds AND @end_ds
-```
-SQLMesh will handle casting the start and end dates to the type of your time column. The format is reflected in the time column format string.
diff --git a/docs/guides/configuration.md b/docs/guides/configuration.md
index 9b345db8..2ebc1aee 100644
--- a/docs/guides/configuration.md
+++ b/docs/guides/configuration.md
@@ -963,7 +963,7 @@ The `VIEW`, `FULL`, and `EMBEDDED` model kinds are specified by name only, while
     );
     ```
 
-    `INCREMENTAL_BY_TIME_RANGE` requires an array specifying the model's `time_column`:
+    `INCREMENTAL_BY_TIME_RANGE` requires an array specifying the model's `time_column` (which should be in the UTC time zone):
 
     ```sql linenums="1"
     MODEL(
diff --git a/docs/guides/incremental_time.md b/docs/guides/incremental_time.md
index f610a528..8878e6e4 100644
--- a/docs/guides/incremental_time.md
+++ b/docs/guides/incremental_time.md
@@ -109,6 +109,10 @@ The model configuration specifies that the column `model_time_column` represents
 
 The `WHERE` clause uses the [SQLMesh predefined macro variables](../concepts/macros/macro_variables.md#predefined-variables) `@start_ds` and `@end_ds` to specify the date range. SQLMesh automatically substitutes in the correct dates based on which intervals are being processed in a job.
 
+!!! tip "Important"
+
+    The `time_column` should be in the [UTC time zone](https://en.wikipedia.org/wiki/Coordinated_Universal_Time) to ensure correct interaction with SQLMesh's scheduler and predefined macro variables.
+
 In addition to the query `WHERE` clause, SQLMesh prevents data leakage by automatically wrapping the query in another time-filtering `WHERE` clause using the time column in the model's configuration.
 
 This raises a question: if SQLMesh automatically adds a time filtering `WHERE` clause, why do you need to include one in the query? Because the two filters play different roles:
diff --git a/docs/reference/model_configuration.md b/docs/reference/model_configuration.md
index a268aa0e..8f1d9b47 100644
--- a/docs/reference/model_configuration.md
+++ b/docs/reference/model_configuration.md
@@ -101,11 +101,11 @@ Configuration options for [`INCREMENTAL_BY_TIME_RANGE` models](../concepts/model
 
 | Option              | Description                                                                                                                                                                                                                                                                                                                      | Type | Required |
 |---------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|:----:|:--------:|
-| `time_column`       | The model column containing each row's timestamp.                                                                                                                                                                                                                                                                                | str  | Y        |
-| `format`            | Argument to `time_column`. Format of the time column's data. (Default: `%Y-%m-%d`)                                                                                                                                                                                                                                               | str  | N        |
-| `batch_size`        | The maximum number of intervals that can be evaluated in a single backfill task. If this is `None`, all intervals will be processed as part of a single task. If this is set, a model's backfill will be chunked such that each individual task only contains jobs with the maximum of `batch_size` intervals. (Default: `None`) | int  | N        |
-| `batch_concurrency` | The maximum number of batches that can run concurrently for this model. (Default: the number of concurrent tasks set in the connection settings)                                                                                                                                                                                 | int  | N        |
-| `lookback`          | The number of time unit intervals prior to the current interval that should be processed. (Default: `0`)                                                                                                                                                                                                                         | int  | N        |
+| `time_column`       | The model column containing each row's timestamp. Should be UTC time zone.                                                                                                                                                                                                                                                       | str  |    Y     |
+| `format`            | Argument to `time_column`. Format of the time column's data. (Default: `%Y-%m-%d`)                                                                                                                                                                                                                                               | str  |    N     |
+| `batch_size`        | The maximum number of intervals that can be evaluated in a single backfill task. If this is `None`, all intervals will be processed as part of a single task. If this is set, a model's backfill will be chunked such that each individual task only contains jobs with the maximum of `batch_size` intervals. (Default: `None`) | int  |    N     |
+| `batch_concurrency` | The maximum number of batches that can run concurrently for this model. (Default: the number of concurrent tasks set in the connection settings)                                                                                                                                                                                 | int  |    N     |
+| `lookback`          | The number of time unit intervals prior to the current interval that should be processed. (Default: `0`)                                                                                                                                                                                                                         | int  |    N     |
 
 Python model kind `name` enum value: [ModelKindName.INCREMENTAL_BY_TIME_RANGE](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#ModelKindName)
 
diff --git a/sqlmesh/core/constants.py b/sqlmesh/core/constants.py
index 5c742d5c..0346ab78 100644
--- a/sqlmesh/core/constants.py
+++ b/sqlmesh/core/constants.py
@@ -1,8 +1,8 @@
 from __future__ import annotations
 
 import datetime
-import multiprocessing as mp
 import os
+import sys
 import typing as t
 from pathlib import Path
 
@@ -32,11 +32,27 @@ MAX_MODEL_DEFINITION_SIZE = 10000
 """Maximum number of characters in a model definition"""
 
 
+def is_daemon_process() -> bool:
+    """
+    Determines whether the current process is running as a daemon (background process).
+
+    This function checks if the standard output (stdout) is connected to a terminal.
+    It does this by calling `sys.stdout.fileno()` to retrieve the file descriptor
+    for the stdout stream and `os.isatty()` to check if this file descriptor is
+    associated with a terminal device.
+
+    Returns:
+        bool: True if the process is running as a daemon (not attached to a terminal),
+              False if the process is running in a terminal (interactive mode).
+    """
+    return not os.isatty(sys.stdout.fileno())
+
+
 # The maximum number of fork processes, used for loading projects
 # None means default to process pool, 1 means don't fork, :N is number of processes
 # Factors in the number of available CPUs even if the process is bound to a subset of them
 # (e.g. via taskset) to avoid oversubscribing the system and causing kill signals
-if hasattr(os, "fork") and not mp.current_process().daemon:
+if hasattr(os, "fork") and not is_daemon_process():
     try:
         MAX_FORK_WORKERS: t.Optional[int] = int(os.getenv("MAX_FORK_WORKERS"))  # type: ignore
     except TypeError:
@@ -81,7 +97,6 @@ SQLMESH_MACRO = "__sqlmesh__macro__"
 SQLMESH_BUILTIN = "__sqlmesh__builtin__"
 SQLMESH_METADATA = "__sqlmesh__metadata__"
 
-
 BUILTIN = "builtin"
 AIRFLOW = "airflow"
 DBT = "dbt"
diff --git a/sqlmesh/core/dialect.py b/sqlmesh/core/dialect.py
index 1a4f134d..1f40dc61 100644
--- a/sqlmesh/core/dialect.py
+++ b/sqlmesh/core/dialect.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import functools
+import logging
 import re
 import sys
 import typing as t
@@ -33,6 +34,8 @@ SQLMESH_MACRO_PREFIX = "@"
 
 TABLES_META = "sqlmesh.tables"
 
+logger = logging.getLogger(__name__)
+
 
 class Model(exp.Expression):
     arg_types = {"expressions": True}
@@ -314,6 +317,15 @@ def _parse_join(
     return macro
 
 
+def _warn_unsupported(self: Parser) -> None:
+    sql = self._find_sql(self._tokens[0], self._tokens[-1])[: self.error_message_context]
+
+    logger.warning(
+        f"'{sql}' could not be semantically understood as it contains unsupported syntax, SQLMesh will treat the command as is. Note that any references to the model's "
+        "underlying physical table can't be resolved in this case, consider using Jinja as explained here https://sqlmesh.readthedocs.io/en/stable/concepts/macros/macro_variables/#audit-only-variables"
+    )
+
+
 def _parse_select(
     self: Parser,
     nested: bool = False,
@@ -579,7 +591,12 @@ def _props_sql(self: Generator, expressions: t.List[exp.Expression]) -> str:
     size = len(expressions)
 
     for i, prop in enumerate(expressions):
-        sql = self.indent(f"{prop.name} {self.sql(prop, 'value')}")
+        value = prop.args.get("value")
+        if prop.name == "when_matched" and isinstance(value, list):
+            output_value = ", ".join(self.sql(v) for v in value)
+        else:
+            output_value = self.sql(prop, "value")
+        sql = self.indent(f"{prop.name} {output_value}")
 
         if i < size - 1:
             sql += ","
@@ -886,6 +903,7 @@ def extend_sqlglot() -> None:
     _override(Parser, _parse_types)
     _override(Parser, _parse_if)
     _override(Parser, _parse_id_var)
+    _override(Parser, _warn_unsupported)
     _override(Snowflake.Parser, _parse_table_parts)
 
 


Test patch:
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index 51217d1a..92534d82 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -3916,7 +3916,7 @@ def test_when_matched_multiple():
     expressions = d.parse(
         """
         MODEL (
-          name db.employees,
+          name @{schema}.employees,
           kind INCREMENTAL_BY_UNIQUE_KEY (
             unique_key name,
             when_matched WHEN MATCHED AND source.x = 1 THEN UPDATE SET target.salary = COALESCE(source.salary, target.salary),
@@ -3933,7 +3933,7 @@ def test_when_matched_multiple():
         "WHEN MATCHED THEN UPDATE SET __MERGE_TARGET__.salary = COALESCE(__MERGE_SOURCE__.salary, __MERGE_TARGET__.salary)",
     ]
 
-    model = load_sql_based_model(expressions, dialect="hive")
+    model = load_sql_based_model(expressions, dialect="hive", variables={"schema": "db"})
     assert len(model.kind.when_matched) == 2
     assert model.kind.when_matched[0].sql() == expected_when_matched[0]
     assert model.kind.when_matched[1].sql() == expected_when_matched[1]