tz_issue_fix

Signed-off-by: Ajay-Satish-01 <ajay.satishpriya@gmail.com>

Author: Ajay-Satish-01

docs/concepts/macros/macro_variables.md

@@ -55,10 +55,14 @@ SQLMesh uses the python [datetime module](https://docs.python.org/3/library/date
 
 !!! tip "Important"
 
-    Predefined variables with a time component always use the [UTC time zone](https://en.wikipedia.org/wiki/Coordinated_Universal_Time).
+    Macro instants such as `@start_dt`, `@end_dt`, `@start_tstz`, and `@end_tstz` are always stored and rendered as [UTC](https://en.wikipedia.org/wiki/Coordinated_Universal_Time) timestamps. During incremental backfill, `@start_ds` and `@end_ds` also use UTC calendar dates derived from those interval boundaries.
 
     Learn more about timezones and incremental models [here](../models/model_kinds.md#timezones).
 
+Relative CLI and API inputs such as `--start "2 weeks ago"` are interpreted using UTC calendar-day boundaries by default. To anchor relative start, end, and execution-time values to a specific timezone (for example, midnight in `America/Los_Angeles`), pass `--time-zone` on supported commands (`plan`, `render`, `evaluate`, `run`, `audit`, `check_intervals`) or set the project-level `time_zone` config. The CLI flag overrides the config value.
+
+When a **day-or-larger** relative start or end (for example, `"1 week ago"`, `"today"`, `"yesterday"`) is parsed with a configured timezone, `@start_tstz` and `@end_tstz` reflect the correct UTC instant and `@start_ds` / `@end_ds` use that timezone's local calendar date in `render`, `evaluate`, and `audit`. Sub-day relatives such as `"2 hours ago"` ignore `--time-zone` and continue to use UTC-relative parsing. Absolute date strings and `@execution_ds` always use UTC calendar dates.
+
 Prefixes:
 
 * start - The inclusive starting interval of a model run

sqlmesh/cli/main.py

@@ -255,6 +255,7 @@ def init(
 @opt.start_time
 @opt.end_time
 @opt.execution_time
+@opt.time_zone
 @opt.expand
 @click.option(
     "--dialect",
@@ -272,6 +273,7 @@ def render(
     start: TimeLike,
     end: TimeLike,
     execution_time: t.Optional[TimeLike] = None,
+    time_zone: t.Optional[str] = None,
     expand: t.Optional[t.Union[bool, t.Iterable[str]]] = None,
     dialect: t.Optional[str] = None,
     no_format: bool = False,
@@ -285,6 +287,7 @@ def render(
         start=start,
         end=end,
         execution_time=execution_time,
+        time_zone=time_zone,
         expand=expand,
     )
 
@@ -310,6 +313,7 @@ def render(
 @opt.start_time
 @opt.end_time
 @opt.execution_time
+@opt.time_zone
 @click.option(
     "--limit",
     type=int,
@@ -324,6 +328,7 @@ def evaluate(
     start: TimeLike,
     end: TimeLike,
     execution_time: t.Optional[TimeLike] = None,
+    time_zone: t.Optional[str] = None,
     limit: t.Optional[int] = None,
 ) -> None:
     """Evaluate a model and return a dataframe with a default limit of 1000."""
@@ -332,6 +337,7 @@ def evaluate(
         start=start,
         end=end,
         execution_time=execution_time,
+        time_zone=time_zone,
         limit=limit,
     )
     if hasattr(df, "show"):
@@ -394,6 +400,7 @@ def diff(ctx: click.Context, environment: t.Optional[str] = None) -> None:
 @opt.start_time
 @opt.end_time
 @opt.execution_time
+@opt.time_zone
 @click.option(
     "--create-from",
     type=str,
@@ -574,6 +581,7 @@ def plan(
 @click.argument("environment", required=False)
 @opt.start_time
 @opt.end_time
+@opt.time_zone
 @click.option("--skip-janitor", is_flag=True, help="Skip the janitor task.")
 @click.option(
     "--ignore-cron",
@@ -796,6 +804,7 @@ def test(
 @opt.start_time
 @opt.end_time
 @opt.execution_time
+@opt.time_zone
 @click.pass_obj
 @error_handler
 @cli_analytics
@@ -805,9 +814,12 @@ def audit(
     start: TimeLike,
     end: TimeLike,
     execution_time: t.Optional[TimeLike] = None,
+    time_zone: t.Optional[str] = None,
 ) -> None:
     """Run audits for the target model(s)."""
-    if not obj.audit(models=models, start=start, end=end, execution_time=execution_time):
+    if not obj.audit(
+        models=models, start=start, end=end, execution_time=execution_time, time_zone=time_zone
+    ):
         exit(1)
 
 
@@ -827,6 +839,7 @@ def audit(
 )
 @opt.start_time
 @opt.end_time
+@opt.time_zone
 @click.pass_context
 @error_handler
 @cli_analytics
@@ -837,6 +850,7 @@ def check_intervals(
     select_model: t.List[str],
     start: TimeLike,
     end: TimeLike,
+    time_zone: t.Optional[str] = None,
 ) -> None:
     """Show missing intervals in an environment, respecting signals."""
     context = ctx.obj
@@ -847,6 +861,7 @@ def check_intervals(
             select_models=select_model,
             start=start,
             end=end,
+            time_zone=time_zone,
         )
     )
 

sqlmesh/cli/options.py

@@ -37,6 +37,11 @@
     help="The execution time (defaults to now).",
 )
 
+time_zone = click.option(
+    "--time-zone",
+    help="IANA timezone for interpreting relative --start, --end, and --execution-time values (e.g. America/Los_Angeles). Defaults to UTC.",
+)
+
 expand = click.option(
     "--expand",
     multiple=True,

sqlmesh/core/config/connection.py

@@ -1806,10 +1806,27 @@ def _static_connection_kwargs(self) -> t.Dict[str, t.Any]:
         from pyspark.conf import SparkConf
         from pyspark.sql import SparkSession
 
+        from sqlmesh.utils.errors import ConfigError
+        from sqlmesh.utils.java import (
+            is_spark_java_supported,
+            java_major_version,
+            spark_java_options,
+        )
+
+        if not is_spark_java_supported():
+            raise ConfigError(
+                f"Spark is not supported on Java {java_major_version() or 'unknown'}. "
+                "Use Java 17 through 23 when running Spark locally."
+            )
+
         spark_config = SparkConf()
-        if self.config:
-            for k, v in self.config.items():
-                spark_config.set(k, v)
+        config = dict(self.config or {})
+        java_options = spark_java_options(config.pop("spark.driver.extraJavaOptions", ""))
+        if java_options:
+            config["spark.driver.extraJavaOptions"] = java_options
+
+        for k, v in config.items():
+            spark_config.set(k, v)
 
         if self.config_dir:
             os.environ["SPARK_CONF_DIR"] = self.config_dir

sqlmesh/core/config/root.py

@@ -46,7 +46,7 @@
 from sqlmesh.core.loader import Loader, SqlMeshLoader
 from sqlmesh.core.notification_target import NotificationTarget
 from sqlmesh.core.user import User
-from sqlmesh.utils.date import to_timestamp, now
+from sqlmesh.utils.date import parse_time_zone, to_timestamp, now
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.pydantic import model_validator
 
@@ -76,16 +76,26 @@ def validate_regex_key_dict(value: t.Dict[str | re.Pattern, t.Any]) -> t.Dict[re
     return compile_regex_mapping(value)
 
 
+def validate_time_zone(v: t.Any) -> t.Optional[str]:
+    if not v or v == "UTC":
+        return None
+    v = str(v)
+    parse_time_zone(v)
+    return v
+
+
 if t.TYPE_CHECKING:
     from sqlmesh.core._typing import Self
 
     NoPastTTLString = str
     GatewayDict = t.Dict[str, GatewayConfig]
     RegexKeyDict = t.Dict[re.Pattern, str]
+    TimeZoneString = t.Optional[str]
 else:
     NoPastTTLString = t.Annotated[str, BeforeValidator(validate_no_past_ttl)]
     GatewayDict = t.Annotated[t.Dict[str, GatewayConfig], BeforeValidator(gateways_ensure_dict)]
     RegexKeyDict = t.Annotated[t.Dict[re.Pattern, str], BeforeValidator(validate_regex_key_dict)]
+    TimeZoneString = t.Annotated[t.Optional[str], BeforeValidator(validate_time_zone)]
 
 
 class Config(BaseConfig):
@@ -129,6 +139,8 @@ class Config(BaseConfig):
         before_all: SQL statements or macros to be executed at the start of the `sqlmesh plan` and `sqlmesh run` commands.
         after_all: SQL statements or macros to be executed at the end of the `sqlmesh plan` and `sqlmesh run` commands.
         cache_dir: The directory to store the SQLMesh cache. Defaults to .cache in the project folder.
+        time_zone: IANA timezone for interpreting relative start, end, and execution-time values.
+            Defaults to UTC when unset.
     """
 
     gateways: GatewayDict = {"": GatewayConfig()}
@@ -174,6 +186,7 @@ class Config(BaseConfig):
     linter: LinterConfig = LinterConfig()
     janitor: JanitorConfig = JanitorConfig()
     cache_dir: t.Optional[str] = None
+    time_zone: TimeZoneString = None
     dbt: t.Optional[DbtConfig] = None
 
     _FIELD_UPDATE_STRATEGY: t.ClassVar[t.Dict[str, UpdateStrategy]] = {