Repository: TobikoData/sqlmesh
Pull request #2295 (2024-03-18 16:17:46)

Docs: add version to list of unsupported dbt methods


Patch:
diff --git a/docs/integrations/dbt.md b/docs/integrations/dbt.md
index 24f4528a..1c2fbe79 100644
--- a/docs/integrations/dbt.md
+++ b/docs/integrations/dbt.md
@@ -263,6 +263,7 @@ The dbt jinja methods that are not currently supported are:
 * schemas
 * graph.nodes.values
 * graph.metrics.values
+* version - learn more about why SQLMesh doesn't support model versions at the [Tobiko Data blog](https://tobikodata.com/the-false-promise-of-dbt-contracts.html)
 
 ## Missing something you need?
 
diff --git a/docs/reference/model_configuration.md b/docs/reference/model_configuration.md
index f5b92997..07c2e552 100644
--- a/docs/reference/model_configuration.md
+++ b/docs/reference/model_configuration.md
@@ -20,6 +20,7 @@ Configuration options for SQLMesh model properties. Supported by all model kinds
 | `cron`             | The cron expression specifying how often the model should be refreshed. (Default: `@daily`)                                                                                                                                                                                                                                      |        str        |    N     |
 | `interval_unit`    | The temporal granularity of the model's data intervals. Supported values: `year`, `month`, `day`, `hour`, `half_hour`, `quarter_hour`, `five_minute`. (Default: inferred from `cron`)                                                                                                                                            |        str        |    N     |
 | `start`            | The date/time that determines the earliest date interval that should be processed by a model. Can be a datetime string, epoch time in milliseconds, or a relative datetime such as `1 year ago`.                                                                                                                                 |    str \| int     |    N     |
+| `end`              | The date/time that determines the latest date interval that should be processed by a model. Can be a datetime string, epoch time in milliseconds, or a relative datetime such as `1 year ago`.                                                                                                                                   |    str \| int     |    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     |
 | `grains`           | The column(s) whose combination uniquely identifies each row in the model                                                                                                                                                                                                                                                        | str \| array[str] |    N     |
 | `references`       | The model column(s) used to join to other models' grains                                                                                                                                                                                                                                                                         | str \| array[str] |    N     |
@@ -43,6 +44,7 @@ The SQLMesh project-level `model_defaults` key supports the following options, d
 - cron
 - owner
 - start
+- end
 - batch_size
 - storage_format
 
diff --git a/setup.py b/setup.py
index 840b155b..1f667964 100644
--- a/setup.py
+++ b/setup.py
@@ -45,7 +45,7 @@ setup(
         "requests",
         "rich[jupyter]",
         "ruamel.yaml",
-        "sqlglot[rs]~=22.5.0",
+        "sqlglot[rs]~=23.0.0",
     ],
     extras_require={
         "bigquery": [
diff --git a/sqlmesh/__init__.py b/sqlmesh/__init__.py
index 8006475a..9324f40d 100644
--- a/sqlmesh/__init__.py
+++ b/sqlmesh/__init__.py
@@ -11,6 +11,7 @@ import sys
 import typing as t
 from datetime import datetime
 from enum import Enum
+from pathlib import Path
 
 from sqlmesh.core.dialect import extend_sqlglot
 
@@ -93,7 +94,7 @@ if RuntimeEnv.get().is_notebook:
 
 
 LOG_FORMAT = "%(asctime)s - %(threadName)s - %(name)s - %(levelname)s - %(message)s (%(filename)s:%(lineno)d)"
-LOG_PREFIX = "logs/sqlmesh_"
+LOG_FILENAME_PREFIX = "sqlmesh_"
 
 
 # SO: https://stackoverflow.com/questions/384076/how-can-i-color-python-logging-output
@@ -126,6 +127,7 @@ def configure_logging(
     write_to_stdout: bool = False,
     write_to_file: bool = True,
     log_limit: int = c.DEFAULT_LOG_LIMIT,
+    log_file_dir: t.Optional[t.Union[str, Path]] = None,
 ) -> None:
     logger = logging.getLogger()
     debug = force_debug or debug_mode_enabled()
@@ -141,9 +143,11 @@ def configure_logging(
     )
     logger.addHandler(stdout_handler)
 
+    log_file_dir = log_file_dir or c.DEFAULT_LOG_FILE_DIR
+    log_path_prefix = Path(log_file_dir) / LOG_FILENAME_PREFIX
     if write_to_file:
-        os.makedirs("logs", exist_ok=True)
-        filename = f"{LOG_PREFIX}{datetime.now().strftime('%Y_%m_%d_%H_%M_%S')}.log"
+        os.makedirs(str(log_file_dir), exist_ok=True)
+        filename = f"{log_path_prefix}{datetime.now().strftime('%Y_%m_%d_%H_%M_%S')}.log"
         file_handler = logging.FileHandler(filename, mode="w", encoding="utf-8")
         # the log files should always log at least info so that users will always have
         # minimal info for debugging even if they specify "ignore_warnings"
@@ -152,7 +156,7 @@ def configure_logging(
         logger.addHandler(file_handler)
 
     if log_limit > 0:
-        for path in list(sorted(glob.glob(f"{LOG_PREFIX}*.log"), reverse=True))[log_limit:]:
+        for path in list(sorted(glob.glob(f"{log_path_prefix}*.log"), reverse=True))[log_limit:]:
             os.remove(path)
 
     if debug:
diff --git a/sqlmesh/cli/main.py b/sqlmesh/cli/main.py
index 02cbf672..6c845c03 100644
--- a/sqlmesh/cli/main.py
+++ b/sqlmesh/cli/main.py
@@ -52,6 +52,11 @@ def _sqlmesh_version() -> str:
     is_flag=True,
     help="Display logs in stdout.",
 )
+@click.option(
+    "--log-file-dir",
+    type=str,
+    help="The directory to write log files to.",
+)
 @click.pass_context
 @error_handler
 def cli(
@@ -62,6 +67,7 @@ def cli(
     ignore_warnings: bool = False,
     debug: bool = False,
     log_to_stdout: bool = False,
+    log_file_dir: t.Optional[str] = None,
 ) -> None:
     """SQLMesh command line tool."""
     if "--help" in sys.argv:
@@ -79,7 +85,9 @@ def cli(
 
     configs = load_configs(config, Context.CONFIG_TYPE, paths)
     log_limit = list(configs.values())[0].log_limit
-    configure_logging(debug, ignore_warnings, log_to_stdout, log_limit=log_limit)
+    configure_logging(
+        debug, ignore_warnings, log_to_stdout, log_limit=log_limit, log_file_dir=log_file_dir
+    )
 
     try:
         context = Context(
diff --git a/sqlmesh/core/constants.py b/sqlmesh/core/constants.py
index f5e1be2c..5440c92a 100644
--- a/sqlmesh/core/constants.py
+++ b/sqlmesh/core/constants.py
@@ -36,6 +36,9 @@ DEFAULT_MAX_LIMIT = 1000
 DEFAULT_LOG_LIMIT = 20
 """The default number of logs to keep."""
 
+DEFAULT_LOG_FILE_DIR = "logs"
+"""The default directory for log files."""
+
 AUDITS = "audits"
 MACROS = "macros"
 METRICS = "metrics"
diff --git a/sqlmesh/core/engine_adapter/base.py b/sqlmesh/core/engine_adapter/base.py
index c7e04c2b..3663525c 100644
--- a/sqlmesh/core/engine_adapter/base.py
+++ b/sqlmesh/core/engine_adapter/base.py
@@ -1015,12 +1015,15 @@ class EngineAdapter:
         if cls.ESCAPE_JSON:
             if isinstance(value, str):
                 return double_escape(value)
-            return value.transform(
-                lambda e: (
-                    exp.Literal.string(double_escape(e.name))
-                    if isinstance(e, exp.Literal) and e.args["is_string"]
-                    else e
-                )
+            return t.cast(
+                exp.Query,
+                value.transform(
+                    lambda e: (
+                        exp.Literal.string(double_escape(e.name))
+                        if isinstance(e, exp.Literal) and e.args["is_string"]
+                        else e
+                    )
+                ),
             )
         return value
 
diff --git a/sqlmesh/core/engine_adapter/mixins.py b/sqlmesh/core/engine_adapter/mixins.py
index 8cd56c28..6aebd6a5 100644
--- a/sqlmesh/core/engine_adapter/mixins.py
+++ b/sqlmesh/core/engine_adapter/mixins.py
@@ -146,28 +146,37 @@ class HiveMetastoreTablePropertiesMixin(EngineAdapter):
             properties.append(exp.FileFormatProperty(this=exp.Var(this=storage_format)))
 
         if partitioned_by:
-            for expr in partitioned_by:
-                if not isinstance(expr, exp.Column):
-                    raise SQLMeshError(
-                        f"PARTITIONED BY contains non-column value '{expr.sql(dialect='spark')}'."
-                    )
-
-            property: exp.Property = exp.PartitionedByProperty(
-                this=exp.Schema(expressions=partitioned_by),
-            )
-
             if (
                 self.dialect == "trino"
                 and self.get_catalog_type(catalog_name or self.get_current_catalog()) == "iceberg"
             ):
                 # On the Trino Iceberg catalog, the table property is called "partitioning" - not "partitioned_by"
                 # In addition, partition column transform expressions like `day(col)` or `bucket(col, 5)` are allowed
+                # Also, column names and transforms need to be strings and supplied as an ARRAY[varchar]
                 # ref: https://trino.io/docs/current/connector/iceberg.html#table-properties
-                property = exp.Property(
-                    this=exp.var("PARTITIONING"), value=exp.Schema(expressions=partitioned_by)
+                properties.append(
+                    exp.Property(
+                        this=exp.var("PARTITIONING"),
+                        value=exp.array(
+                            *(
+                                exp.Literal.string(e.sql(dialect=self.dialect))
+                                for e in partitioned_by
+                            )
+                        ),
+                    )
+                )
+            else:
+                for expr in partitioned_by:
+                    if not isinstance(expr, exp.Column):
+                        raise SQLMeshError(
+                            f"PARTITIONED BY contains non-column value '{expr.sql(dialect=self.dialect)}'."
+                        )
+
+                properties.append(
+                    exp.PartitionedByProperty(
+                        this=exp.Schema(expressions=partitioned_by),
+                    )
                 )
-
-            properties.append(property)
 
         if table_description:
             properties.append(
diff --git a/sqlmesh/core/engine_adapter/shared.py b/sqlmesh/core/engine_adapter/shared.py
index 4c299e8e..b6b0e328 100644
--- a/sqlmesh/core/engine_adapter/shared.py
+++ b/sqlmesh/core/engine_adapter/shared.py
@@ -236,7 +236,7 @@ class SourceQuery:
     def __enter__(self) -> Query:
         query = self.query_factory()
         for transform in self._transforms:
-            query = query.transform(transform)
+            query = t.cast(exp.Query, query.transform(transform))
         return query
 
     def __exit__(
diff --git a/sqlmesh/core/loader.py b/sqlmesh/core/loader.py
index 73cfbdc2..54144c7e 100644
--- a/sqlmesh/core/loader.py
+++ b/sqlmesh/core/loader.py
@@ -156,7 +156,7 @@ class Loader(abc.ABC):
             True if a modification is found; False otherwise
         """
         return any(
-            path.stat().st_mtime > initial_mtime
+            not path.exists() or path.stat().st_mtime > initial_mtime
             for path, initial_mtime in self._path_mtimes.items()
         )
 
diff --git a/sqlmesh/core/macros.py b/sqlmesh/core/macros.py
index 6424400b..ed51936a 100644
--- a/sqlmesh/core/macros.py
+++ b/sqlmesh/core/macros.py
@@ -168,9 +168,6 @@ class MacroEvaluator:
         ) -> exp.Expression | t.List[exp.Expression] | None:
             nonlocal changed
 
-            exp.replace_children(
-                node, lambda n: n if isinstance(n, exp.Lambda) else evaluate_macros(n)
-            )
             if isinstance(node, MacroVar):
                 changed = True
                 if node.name not in self.locals:
@@ -198,7 +195,9 @@ class MacroEvaluator:
                 return self.evaluate(node)
             return node
 
-        transformed = evaluate_macros(expression.copy())
+        transformed = exp.replace_tree(
+            expression.copy(), evaluate_macros, prune=lambda n: isinstance(n, exp.Lambda)
+        )
 
         if changed:
             # the transformations could have corrupted the ast, turning this into sql and reparsing ensures
@@ -230,7 +229,7 @@ class MacroEvaluator:
             if isinstance(node.expression, exp.Lambda):
                 _, fn = _norm_var_arg_lambda(self, node.expression)
                 self.macros[normalize_macro_name(node.name)] = lambda _, *args: fn(
-                    list(args) if len(args) <= 1 else exp.Tuple(expressions=list(args))
+                    args[0] if len(args) == 1 else exp.Tuple(expressions=list(args))
                 )
             else:
                 self.locals[node.name] = self.transform(node.expression)
diff --git a/sqlmesh/core/metric/definition.py b/sqlmesh/core/metric/definition.py
index 0c968672..10da247b 100644
--- a/sqlmesh/core/metric/definition.py
+++ b/sqlmesh/core/metric/definition.py
@@ -111,7 +111,7 @@ class MetricMeta(PydanticModel, frozen=True):
         metric_refs = {}
         agg_or_ref = False
 
-        for node, *_ in self.expression.walk():
+        for node in self.expression.walk():
             if isinstance(node, exp.Alias):
                 _raise_metric_config_error(
                     f"Alias found for metric '{self.name}' which is not allowed", self._path
@@ -157,12 +157,15 @@ class Metric(MetricMeta, frozen=True):
         This method removes catalog and schema information from columns.
         """
         return {
-            t.cast(exp.Expression, agg.parent).transform(
-                lambda node: (
-                    exp.column(node.this, table=remove_namespace(node))
-                    if isinstance(node, exp.Column) and node.table
-                    else node
-                )
+            t.cast(
+                exp.AggFunc,
+                t.cast(exp.Expression, agg.parent).transform(
+                    lambda node: (
+                        exp.column(node.this, table=remove_namespace(node))
+                        if isinstance(node, exp.Column) and node.table
+                        else node
+                    )
+                ),
             ): _get_measure_and_dim_tables(agg)
             for agg in self.expanded.find_all(exp.AggFunc)
         }
@@ -207,7 +210,7 @@ def _get_measure_and_dim_tables(expression: exp.Expression) -> MeasureAndDimTabl
             return is_measure(parent)
         return False
 
-    for node, _, key in expression.walk():
+    for node in expression.walk():
         if isinstance(node, exp.Column) and node.table:
             table = ".".join(p.sql() for p in node.parts[:-1])
             tables[table] = True
diff --git a/sqlmesh/core/metric/rewriter.py b/sqlmesh/core/metric/rewriter.py
index 3f9f85df..3519a77e 100644
--- a/sqlmesh/core/metric/rewriter.py
+++ b/sqlmesh/core/metric/rewriter.py
@@ -131,7 +131,7 @@ class Rewriter:
         mapping = {v: k for k, v in mapping.items()}
 
         for expr in grain:
-            for node, *_ in expr.walk():
+            for node in expr.walk():
                 if isinstance(node, exp.Column):
                     models = self.graph.models_for_column(name, node.name)
 
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index 11ae7598..9b2b9a7d 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -101,6 +101,8 @@ class _Model(ModelMeta, frozen=True):
         start: The earliest date that the model will be backfilled for. If this is None,
             then the date is inferred by taking the most recent start date of its ancestors.
             The start date can be a static datetime or a relative datetime like "1 year ago"
+        end: The date that the model will be backfilled up until. Follows the same syntax as 'start',
+            should be omitted if there is no end date.
         batch_size: The maximum number of incremental intervals that can be run per backfill job. If this is None,
             then backfilling this model will do all of history in one job. If this is set, a model's backfill
             will be chunked such that each individual job will only contain jobs with max `batch_size` intervals.
diff --git a/sqlmesh/core/model/kind.py b/sqlmesh/core/model/kind.py
index 4f9b6cae..630ec9d4 100644
--- a/sqlmesh/core/model/kind.py
+++ b/sqlmesh/core/model/kind.py
@@ -112,7 +112,7 @@ class ModelKindMixin:
     @property
     def only_execution_time(self) -> bool:
         """Whether or not this model only cares about execution time to render."""
-        return self.is_view or self.is_full or self.is_incremental_unmanaged or self.is_scd_type_2
+        return self.is_view or self.is_full or self.is_scd_type_2
 
 
 class ModelKindName(str, ModelKindMixin, Enum):
@@ -301,7 +301,7 @@ class IncrementalByUniqueKeyKind(_Incremental):
         if not v:
             return v
 
-        return v.transform(replace_table_references)
+        return t.cast(exp.When, v.transform(replace_table_references))
 
 
 class IncrementalUnmanagedKind(_ModelKind):
diff --git a/sqlmesh/core/node.py b/sqlmesh/core/node.py
index 50df3efa..080b5619 100644
--- a/sqlmesh/core/node.py
+++ b/sqlmesh/core/node.py
@@ -9,7 +9,7 @@ from pydantic import Field
 from sqlglot import exp
 
 from sqlmesh.utils.cron import CroniterCache
-from sqlmesh.utils.date import TimeLike, to_datetime
+from sqlmesh.utils.date import TimeLike, to_datetime, validate_date_range
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.pydantic import (
     PydanticModel,
@@ -172,6 +172,8 @@ class _Node(PydanticModel):
         start: The earliest date that the node will be executed for. If this is None,
             then the date is inferred by taking the most recent start date of its ancestors.
             The start date can be a static datetime or a relative datetime like "1 year ago"
+        end: The latest date that the model will be executed for. If this is None,
+            the date from the scheduler will be used
         cron: A cron string specifying how often the node should be run, leveraging the
             [croniter](https://github.com/kiorky/croniter) library.
         interval_unit: The duration of an interval for the node. By default, it is computed from the cron expression.
@@ -185,6 +187,7 @@ class _Node(PydanticModel):
     description: t.Optional[str] = None
     owner: t.Optional[str] = None
     start: t.Optional[TimeLike] = None
+    end: t.Optional[TimeLike] = None
     cron: str = "@daily"
     interval_unit_: t.Optional[IntervalUnit] = Field(alias="interval_unit", default=None)
     tags: t.List[str] = []
@@ -207,7 +210,7 @@ class _Node(PydanticModel):
             return v.meta["sql"]
         return str(v)
 
-    @field_validator("start", mode="before")
+    @field_validator("start", "end", mode="before")
     @classmethod
     def _date_validator(cls, v: t.Any) -> t.Optional[TimeLike]:
         if isinstance(v, exp.Expression):
@@ -255,6 +258,7 @@ class _Node(PydanticModel):
                 raise ConfigError(
                     f"Interval unit of '{interval_unit}' is larger than cron period of '{cron}'"
                 )
+        validate_date_range(values.get("start"), values.get("end"))
         return values
 
     @property
diff --git a/sqlmesh/core/renderer.py b/sqlmesh/core/renderer.py
index 05e9f4ba..1e0b2606 100644
--- a/sqlmesh/core/renderer.py
+++ b/sqlmesh/core/renderer.py
@@ -271,7 +271,7 @@ class BaseExpressionRenderer:
                                 logger.warning("Failed to expand the nested model '%s'", name)
                     return node
 
-                expression = expression.transform(_expand, copy=False)
+                expression = expression.transform(_expand, copy=False)  # type: ignore
 
             if mapping:
                 expression = exp.replace_tables(
diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py
index fb330619..f63d5731 100644
--- a/sqlmesh/core/snapshot/definition.py
+++ b/sqlmesh/core/snapshot/definition.py
@@ -1482,12 +1482,17 @@ def missing_intervals(
             snapshot.intervals = snapshot.intervals.copy()
             snapshot.remove_interval(interval, execution_time)
 
+        missing_interval_end_date = snapshot_end_date
+        node_end_date = snapshot.node.end
+        if node_end_date and (to_datetime(node_end_date) < to_datetime(snapshot_end_date)):
+            missing_interval_end_date = node_end_date
+
         intervals = snapshot.missing_intervals(
             max(
                 to_datetime(snapshot_start_date),
                 to_datetime(start_date(snapshot, snapshots, cache, relative_to=snapshot_end_date)),
             ),
-            snapshot_end_date,
+            missing_interval_end_date,
             execution_time=execution_time,
             deployability_index=deployability_index,
             ignore_cron=ignore_cron,
diff --git a/sqlmesh/magics.py b/sqlmesh/magics.py
index 74ae9621..9952ec4d 100644
--- a/sqlmesh/magics.py
+++ b/sqlmesh/magics.py
@@ -91,6 +91,7 @@ class SQLMeshMagics(Magics):
     @argument("--gateway", type=str, help="The name of the gateway.")
     @argument("--ignore-warnings", action="store_true", help="Ignore warnings.")
     @argument("--debug", action="store_true", help="Enable debug mode.")
+    @argument("--log-file-dir", type=str, help="The directory to write the log file to.")
     @line_magic
     def context(self, line: str) -> None:
         """Sets the context in the user namespace."""
@@ -99,7 +100,9 @@ class SQLMeshMagics(Magics):
         args = parse_argstring(self.context, line)
         configs = load_configs(args.config, Context.CONFIG_TYPE, args.paths)
         log_limit = list(configs.values())[0].log_limit
-        configure_logging(args.debug, args.ignore_warnings, log_limit=log_limit)
+        configure_logging(
+            args.debug, args.ignore_warnings, log_limit=log_limit, log_file_dir=args.log_file_dir
+        )
         try:
             context = Context(paths=args.paths, config=configs, gateway=args.gateway)
             self._shell.user_ns["context"] = context
diff --git a/sqlmesh/schedulers/airflow/dag_generator.py b/sqlmesh/schedulers/airflow/dag_generator.py
index 3f261113..a2c69dd9 100644
--- a/sqlmesh/schedulers/airflow/dag_generator.py
+++ b/sqlmesh/schedulers/airflow/dag_generator.py
@@ -108,10 +108,15 @@ class SnapshotDagGenerator:
                 f"Can't create a cadence DAG for the paused snapshot {snapshot.snapshot_id}"
             )
 
+        end_date = None
+        if snapshot.node.end:
+            end_date = pendulum.instance(to_datetime(snapshot.node.end))
+
         with DAG(
             dag_id=dag_id,
             schedule_interval=snapshot.node.cron,
             start_date=pendulum.instance(to_datetime(snapshot.unpaused_ts)),
+            end_date=end_date,
             max_active_runs=1,
             catchup=True,
             is_paused_upon_creation=False,
diff --git a/web/client/src/library/components/documentation/Documentation.tsx b/web/client/src/library/components/documentation/Documentation.tsx
index 2a574fd2..b093dad6 100644
--- a/web/client/src/library/components/documentation/Documentation.tsx
+++ b/web/client/src/library/components/documentation/Documentation.tsx
@@ -47,7 +47,7 @@ const Documentation = function Documentation({
     return () => {
       cancelRequestModel()
     }
-  }, [model.hash])
+  }, [model.name, model.hash])
 
   return (
     <Container>
diff --git a/web/client/src/library/components/editor/EditorInspector.tsx b/web/client/src/library/components/editor/EditorInspector.tsx
index 255470a7..18f79a03 100644
--- a/web/client/src/library/components/editor/EditorInspector.tsx
+++ b/web/client/src/library/components/editor/EditorInspector.tsx
@@ -584,7 +584,7 @@ function FormDiffModel({
     void getDiff().then(({ data }) => {
       setPreviewDiff(data)
     })
-  }, [model.hash])
+  }, [model.name, model.hash])
 
   useEffect(() => {
     return () => {
diff --git a/web/client/src/library/components/fileExplorer/Directory.tsx b/web/client/src/library/components/fileExplorer/Directory.tsx
index 8aada818..01245627 100644
--- a/web/client/src/library/components/fileExplorer/Directory.tsx
+++ b/web/client/src/library/components/fileExplorer/Directory.tsx
@@ -32,7 +32,7 @@ const Directory = function Directory({
   const [isDraggable, setIsDraggable] = useState(false)
 
   const attrs = useLongPress(() => setIsDraggable(true), {
-    threshold: 500,
+    threshold: 50,
     onFinish() {
       setIsDraggable(false)
     },
diff --git a/web/client/src/library/components/fileExplorer/File.tsx b/web/client/src/library/components/fileExplorer/File.tsx
index 632e0e20..13ad0f7f 100644
--- a/web/client/src/library/components/fileExplorer/File.tsx
+++ b/web/client/src/library/components/fileExplorer/File.tsx
@@ -35,7 +35,7 @@ function File({
   const [isDraggable, setIsDraggable] = useState(false)
 
   const attrs = useLongPress(() => setIsDraggable(true), {
-    threshold: 500,
+    threshold: 50,
     onFinish() {
       setIsDraggable(false)
     },
diff --git a/web/client/src/library/components/graph/ModelLineage.tsx b/web/client/src/library/components/graph/ModelLineage.tsx
index cd50bf3c..3e5d8ab1 100644
--- a/web/client/src/library/components/graph/ModelLineage.tsx
+++ b/web/client/src/library/components/graph/ModelLineage.tsx
@@ -118,7 +118,7 @@ export default function ModelLineage({
       setMainNode(undefined)
       setHighlightedNodes({})
     }
-  }, [model.hash])
+  }, [model.name, model.hash])
 
   useEffect(() => {
     Object.keys(modelLineage ?? {}).forEach(modelName => {
diff --git a/web/client/src/library/components/loading/LoadingStatus.tsx b/web/client/src/library/components/loading/LoadingStatus.tsx
index cb62a2dc..516f9605 100644
--- a/web/client/src/library/components/loading/LoadingStatus.tsx
+++ b/web/client/src/library/components/loading/LoadingStatus.tsx
@@ -9,7 +9,7 @@ export default function LoadingStatus({
   return (
     <Loading className="inline-block">
       <Spinner className="w-3 h-3 border border-neutral-10 mr-2" />
-      <span className="inline-block text-xs">{children}</span>
+      <span className="inline-block text-xs whitespace-nowrap">{children}</span>
     </Loading>
   )
 }
diff --git a/web/client/src/library/pages/root/Navigation.tsx b/web/client/src/library/pages/root/Navigation.tsx
index a1688d8d..84116c3b 100644
--- a/web/client/src/library/pages/root/Navigation.tsx
+++ b/web/client/src/library/pages/root/Navigation.tsx
@@ -14,16 +14,17 @@ export default function PageNavigation(): JSX.Element {
   return (
     <div
       className={clsx(
-        'min-w-[10rem] px-2 min-h-8 max-h-8 w-full flex items-center',
+        'relative min-w-[10rem] px-2 min-h-8 max-h-8 w-full flex items-center',
         modules.showHistoryNavigation ? 'justify-between' : 'justify-end',
       )}
     >
       {modules.showHistoryNavigation && <HistoryNavigation />}
-      {isFetchingModels ? (
-        <LoadingStatus>Loading Models...</LoadingStatus>
-      ) : (
-        modules.hasPlans && <EnvironmentDetails />
+      {isFetchingModels && (
+        <div className="absolute w-full h-full flex justify-center items-center z-10 bg-transparent-20 backdrop-blur-lg">
+          <LoadingStatus>Loading Models...</LoadingStatus>
+        </div>
       )}
+      {modules.hasPlans && <EnvironmentDetails />}
       {modules.hasErrors && <ReportErrors />}
     </div>
   )
diff --git a/web/server/api/endpoints/directories.py b/web/server/api/endpoints/directories.py
index e9595fee..11fa2f44 100644
--- a/web/server/api/endpoints/directories.py
+++ b/web/server/api/endpoints/directories.py
@@ -5,10 +5,9 @@ import typing as t
 from fastapi import APIRouter, Body, Depends, HTTPException, Response, status
 from starlette.status import HTTP_404_NOT_FOUND
 
-from sqlmesh.core.context import Context
 from web.server import models
 from web.server.exceptions import ApiException
-from web.server.settings import Settings, get_context, get_settings
+from web.server.settings import Settings, get_settings
 from web.server.utils import replace_file, validate_path
 
 router = APIRouter()
@@ -16,15 +15,13 @@ router = APIRouter()
 
 @router.post("/{path:path}", response_model=models.Directory)
 async def write_directory(
-    response: Response,
     path: str = Depends(validate_path),
     new_path: t.Optional[str] = Body(None, embed=True),
     settings: Settings = Depends(get_settings),
-    context: Context = Depends(get_context),
 ) -> models.Directory:
     """Create or rename a directory."""
     if new_path:
-        validate_path(new_path, context)
+        new_path = await validate_path(new_path, settings)
         replace_file(settings.project_path / path, settings.project_path / new_path)
         return models.Directory(name=os.path.basename(new_path), path=new_path)
 
diff --git a/web/server/api/endpoints/files.py b/web/server/api/endpoints/files.py
index 9a1f0bba..56e7bafe 100644
--- a/web/server/api/endpoints/files.py
+++ b/web/server/api/endpoints/files.py
@@ -25,16 +25,9 @@ router = APIRouter()
 
 
 @router.get("", response_model=models.Directory)
-def get_files(
-    context: t.Optional[Context] = Depends(get_context),
-    settings: Settings = Depends(get_settings),
-) -> models.Directory:
+async def get_files(settings: Settings = Depends(get_settings)) -> models.Directory:
     """Get all project files."""
-    return _get_directory(
-        path=settings.project_path,
-        settings=settings,
-        context=context,
-    )
+    return await _get_directory(settings.project_path, settings)
 
 
 @router.get("/{path:path}", response_model=models.File)
@@ -54,21 +47,26 @@ def get_file(
 @router.post("/{path:path}", response_model=t.Optional[models.File])
 async def write_file(
     response: Response,
+    path: str = Depends(validate_path),
     content: str = Body("", embed=True),
     new_path: t.Optional[str] = Body(None, embed=True),
-    path: str = Depends(validate_path),
     settings: Settings = Depends(get_settings),
-    context: Context = Depends(get_context),
+    context: t.Optional[Context] = Depends(get_context),
 ) -> t.Optional[models.File]:
     """Create, update, or rename a file."""
     path_or_new_path = path
     if new_path:
-        path_or_new_path = validate_path(new_path, context)
+        path_or_new_path = await validate_path(new_path, settings)
         replace_file(settings.project_path / path, settings.project_path / path_or_new_path)
     else:
         full_path = settings.project_path / path
-        config = context.config_for_path(Path(path_or_new_path))
-        if config.ui.format_on_save and content and Path(path_or_new_path).suffix == ".sql":
+        config = context.config_for_path(Path(path_or_new_path)) if context else None
+        if (
+            config
+            and config.ui.format_on_save
+            and content
+            and Path(path_or_new_path).suffix == ".sql"
+        ):
             format_file_status = models.FormatFileStatus(
                 status=models.Status.INIT, path=path_or_new_path
             )
@@ -120,11 +118,8 @@ async def delete_file(
         )
 
 
-def _get_directory(
-    path: str | Path,
-    settings: Settings,
-    context: t.Optional[Context] = None,
-) -> models.Directory:
+async def _get_directory(path: str | Path, settings: Settings) -> models.Directory:
+    context = await get_context(settings)
     ignore_patterns = context.config.ignore_patterns if context else c.IGNORE_PATTERNS
 
     def walk_path(
diff --git a/web/server/settings.py b/web/server/settings.py
index de20ba97..411d06e0 100644
--- a/web/server/settings.py
+++ b/web/server/settings.py
@@ -125,3 +125,9 @@ async def get_context_or_raise(settings: Settings = Depends(get_settings)) -> Co
             origin="API -> settings -> get_context",
         )
     return context
+
+
+def invalidate_context_cache() -> None:
+    _get_context.cache_clear()
+    _get_loaded_context.cache_clear()
+    _get_path_to_model_mapping.cache_clear()
diff --git a/web/server/utils.py b/web/server/utils.py
index 0a239777..f98d57d2 100644
--- a/web/server/utils.py
+++ b/web/server/utils.py
@@ -12,10 +12,10 @@ from fastapi import Depends, HTTPException
 from starlette.responses import StreamingResponse
 from starlette.status import HTTP_404_NOT_FOUND, HTTP_422_UNPROCESSABLE_ENTITY
 
-from sqlmesh.core.context import Context
+from sqlmesh.core import constants as c
 from web.server.console import api_console
 from web.server.exceptions import ApiException
-from web.server.settings import get_context
+from web.server.settings import Settings, get_context, get_settings
 
 R = t.TypeVar("R")
 
@@ -51,11 +51,9 @@ async def run_in_executor(func: t.Callable[..., R], *args: t.Any) -> R:
     return await loop.run_in_executor(None, func_wrapper)
 
 
-def validate_path(
-    path: str,
-    context: Context = Depends(get_context),
-) -> str:
-    resolved_path = context.path.resolve()
+async def validate_path(path: str, settings: Settings = Depends(get_settings)) -> str:
+    context = await get_context(settings)
+    resolved_path = settings.project_path.resolve()
     full_path = (resolved_path / path).resolve()
     try:
         full_path.relative_to(resolved_path)
@@ -63,7 +61,10 @@ def validate_path(
         raise HTTPException(status_code=HTTP_404_NOT_FOUND)
 
     if any(
-        full_path.match(pattern) for pattern in context.config_for_path(Path(path)).ignore_patterns
+        full_path.match(pattern)
+        for pattern in (
+            context.config_for_path(Path(path)).ignore_patterns if context else c.IGNORE_PATTERNS
+        )
     ):
         raise HTTPException(status_code=HTTP_404_NOT_FOUND)
 
diff --git a/web/server/watcher.py b/web/server/watcher.py
index 748d1a0b..85ec364f 100644
--- a/web/server/watcher.py
+++ b/web/server/watcher.py
@@ -4,34 +4,48 @@ from pathlib import Path
 from watchfiles import Change, DefaultFilter, awatch
 
 from sqlmesh.core import constants as c
+from sqlmesh.core.context import Context
 from web.server import models
 from web.server.api.endpoints.files import _get_directory, _get_file_with_content
 from web.server.console import api_console
 from web.server.exceptions import ApiException
-from web.server.settings import get_context_or_raise, get_settings
+from web.server.settings import (
+    Settings,
+    get_context,
+    get_settings,
+    invalidate_context_cache,
+)
+from web.server.utils import is_relative_to
 
 
 async def watch_project() -> None:
     settings = get_settings()
-    context = await get_context_or_raise(settings)
+    context = await get_context(settings)
+    paths = [
+        (settings.project_path / c.AUDITS).resolve(),
+        (settings.project_path / c.MACROS).resolve(),
+        (settings.project_path / c.MODELS).resolve(),
+        (settings.project_path / c.METRICS).resolve(),
+        (settings.project_path / c.SEEDS).resolve(),
+    ]
     ignore_entity_patterns = context.config.ignore_patterns if context else c.IGNORE_PATTERNS
     ignore_entity_patterns.append("^\\.DS_Store$")
     ignore_entity_patterns.append("^.*\.db(\.wal)?$")
-    ignore_paths = [str((context.path / c.CACHE).resolve())]
+    ignore_paths = [str((settings.project_path / c.CACHE).resolve())]
     watch_filter = DefaultFilter(
         ignore_paths=ignore_paths, ignore_entity_patterns=ignore_entity_patterns
     )
-
-    async for entries in awatch(context.path, watch_filter=watch_filter, force_polling=True):
+    async for entries in awatch(
+        settings.project_path, watch_filter=watch_filter, force_polling=True
+    ):
         changes: t.List[models.ArtifactChange] = []
         directories: t.Dict[str, models.Directory] = {}
         for change, path_str in entries:
             path = Path(path_str)
             try:
                 relative_path = path.relative_to(settings.project_path)
-
                 if change == Change.modified and path.is_dir():
-                    directory = _get_directory(path, settings, context)
+                    directory = await _get_directory(path, settings)
                     directories[directory.path] = directory
                 elif change == Change.deleted or not path.exists():
                     changes.append(
@@ -51,6 +65,16 @@ async def watch_project() -> None:
                             ),
                         )
                     )
+                if context:
+                    in_paths = any(is_relative_to(path, p) for p in paths)
+                    is_modified_new_file = (
+                        change == Change.modified and path not in context._loader._path_mtimes
+                    )
+                    should_track_file = path.is_file() and in_paths
+                    should_reset_mtime = Change.added or is_modified_new_file
+                    if should_track_file and should_reset_mtime:
+                        context._loader._path_mtimes[path] = 0
+
             except Exception:
                 error = ApiException(
                     message="Error updating file",
@@ -64,3 +88,26 @@ async def watch_project() -> None:
                 event=models.EventName.FILE,
                 data={"changes": changes, "directories": directories},
             )
+
+        if is_config_changed(entries, settings, context):
+            invalidate_context_cache()
+            api_console.log_event(
+                event=models.EventName.WARNINGS,
+                data=ApiException(
+                    message="Config file changed",
+                    origin="API -> watcher -> watch_project",
+                    trigger="config",
+                ).to_dict(),
+            )
+
+
+def is_config_changed(
+    entries: t.Set[t.Any], settings: Settings, context: t.Optional[Context] = None
+) -> bool:
+    config_paths = set(context.configs) if context else {settings.project_path}
+
+    return any(
+        Path(path) in config_path.glob("config.*")
+        for config_path in config_paths
+        for _, path in entries
+    )


Test patch:
diff --git a/sqlmesh/core/test/definition.py b/sqlmesh/core/test/definition.py
index ae950dd0..9da8a931 100644
--- a/sqlmesh/core/test/definition.py
+++ b/sqlmesh/core/test/definition.py
@@ -129,20 +129,15 @@ class ModelTest(unittest.TestCase):
         expected = expected.replace({None: np.nan})
 
         def _to_hashable(x: t.Any) -> t.Any:
-            return tuple(x) if isinstance(x, list) else x
+            if isinstance(x, (list, np.ndarray)):
+                return tuple(x)
+            return str(x) if not isinstance(x, t.Hashable) else x
 
         if sort:
-            actual = (
-                actual.apply(_to_hashable)
-                .sort_values(by=actual.columns.to_list())
-                .reset_index(drop=True)
-            )
-            expected = (
-                expected.apply(_to_hashable)
-                .sort_values(by=expected.columns.to_list())
-                .reset_index(drop=True)
-            )
-
+            actual = actual.apply(lambda col: col.map(_to_hashable))
+            actual = actual.sort_values(by=actual.columns.to_list()).reset_index(drop=True)
+            expected = expected.apply(lambda col: col.map(_to_hashable))
+            expected = expected.sort_values(by=expected.columns.to_list()).reset_index(drop=True)
         try:
             pd.testing.assert_frame_equal(
                 expected,
diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py
index b8cfdf8f..b416f193 100644
--- a/tests/cli/test_cli.py
+++ b/tests/cli/test_cli.py
@@ -91,7 +91,9 @@ GROUP BY item_id
 
 
 def init_prod_and_backfill(runner, temp_dir) -> None:
-    result = runner.invoke(cli, ["--paths", temp_dir, "plan", "--auto-apply"])
+    result = runner.invoke(
+        cli, ["--log-file-dir", temp_dir, "--paths", temp_dir, "plan", "--auto-apply"]
+    )
     assert_plan_success(result)
     assert path.exists(temp_dir / "db.db")
 
@@ -133,17 +135,17 @@ def assert_virtual_update(result) -> None:
     assert "Virtual Update executed successfully" in result.output
 
 
-def test_version(runner):
+def test_version(runner, tmp_path):
     from sqlmesh import __version__ as SQLMESH_VERSION
 
-    result = runner.invoke(cli, ["--version"])
+    result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--version"])
     assert result.exit_code == 0
     assert SQLMESH_VERSION in result.output
 
 
 def test_plan_no_config(runner, tmp_path):
     # Error if no SQLMesh project config is found
-    result = runner.invoke(cli, ["--paths", tmp_path, "plan"])
+    result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan"])
     assert result.exit_code == 1
     assert "Error: SQLMesh project config could not be found" in result.output
 
@@ -154,7 +156,9 @@ def test_plan(runner, tmp_path):
     # Example project models have start dates, so there are no date prompts
     # for the `prod` environment.
     # Input: `y` to apply and backfill
-    result = runner.invoke(cli, ["--paths", tmp_path, "plan"], input="y\n")
+    result = runner.invoke(
+        cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan"], input="y\n"
+    )
     assert_plan_success(result)
 
 
@@ -163,7 +167,9 @@ def test_plan_skip_tests(runner, tmp_path):
 
     # Successful test run message should not appear with `--skip-tests`
     # Input: `y` to apply and backfill
-    result = runner.invoke(cli, ["--paths", tmp_path, "plan", "--skip-tests"], input="y\n")
+    result = runner.invoke(
+        cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "--skip-tests"], input="y\n"
+    )
     assert result.exit_code == 0
     assert "Successfully Ran 1 tests against duckdb" not in result.output
     assert_new_env(result)
@@ -178,7 +184,15 @@ def test_plan_restate_model(runner, tmp_path):
     # Input: enter for backfill start date prompt, enter for end date prompt, `y` to apply and backfill
     result = runner.invoke(
         cli,
-        ["--paths", tmp_path, "plan", "--restate-model", "sqlmesh_example.full_model"],
+        [
+            "--log-file-dir",
+            tmp_path,
+            "--paths",
+            tmp_path,
+            "plan",
+            "--restate-model",
+            "sqlmesh_example.full_model",
+        ],
         input="\n\ny\n",
     )
     assert result.exit_code == 0
@@ -192,7 +206,9 @@ def test_plan_skip_backfill(runner, tmp_path):
     create_example_project(tmp_path)
 
     # plan for `prod` errors if `--skip-backfill` is passed without --no-gaps
-    result = runner.invoke(cli, ["--paths", tmp_path, "plan", "--skip-backfill"])
+    result = runner.invoke(
+        cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "--skip-backfill"]
+    )
     assert result.exit_code == 1
     assert (
         "Error: When targeting the production environment either the backfill should not be skipped or the lack of data gaps should be enforced (--no-gaps flag)."
@@ -202,7 +218,9 @@ def test_plan_skip_backfill(runner, tmp_path):
     # plan executes virtual update without executing model batches
     # Input: `y` to perform virtual update
     result = runner.invoke(
-        cli, ["--paths", tmp_path, "plan", "--skip-backfill", "--no-gaps"], input="y\n"
+        cli,
+        ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "--skip-backfill", "--no-gaps"],
+        input="y\n",
     )
     assert result.exit_code == 0
     assert_virtual_update(result)
@@ -213,7 +231,9 @@ def test_plan_auto_apply(runner, tmp_path):
     create_example_project(tmp_path)
 
     # plan for `prod` runs end-to-end with no user input with `--auto-apply`
-    result = runner.invoke(cli, ["--paths", tmp_path, "plan", "--auto-apply"])
+    result = runner.invoke(
+        cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "--auto-apply"]
+    )
     assert_plan_success(result)
 
     # confirm verbose output not present
@@ -225,7 +245,9 @@ def test_plan_verbose(runner, tmp_path):
     create_example_project(tmp_path)
 
     # Input: `y` to apply and backfill
-    result = runner.invoke(cli, ["--paths", tmp_path, "plan", "--verbose"], input="y\n")
+    result = runner.invoke(
+        cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "--verbose"], input="y\n"
+    )
     assert_plan_success(result)
     assert "sqlmesh_example.seed_model created" in result.output
     assert "sqlmesh_example.seed_model promoted" in result.output
@@ -235,7 +257,9 @@ def test_plan_dev(runner, tmp_path):
     create_example_project(tmp_path)
 
     # Input: enter for backfill start date prompt, enter for end date prompt, `y` to apply and backfill
-    result = runner.invoke(cli, ["--paths", tmp_path, "plan", "dev"], input="\n\ny\n")
+    result = runner.invoke(
+        cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "dev"], input="\n\ny\n"
+    )
     assert_plan_success(result, "dev")
 
 
@@ -244,7 +268,9 @@ def test_plan_dev_start_date(runner, tmp_path):
 
     # Input: enter for backfill end date prompt, `y` to apply and backfill
     result = runner.invoke(
-        cli, ["--paths", tmp_path, "plan", "dev", "--start", "2023-01-01"], input="\ny\n"
+        cli,
+        ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "dev", "--start", "2023-01-01"],
+        input="\ny\n",
     )
     assert_plan_success(result, "dev")
     assert "sqlmesh_example__dev.full_model: 2023-01-01" in result.output
@@ -256,7 +282,9 @@ def test_plan_dev_end_date(runner, tmp_path):
 
     # Input: enter for backfill start date prompt, `y` to apply and backfill
     result = runner.invoke(
-        cli, ["--paths", tmp_path, "plan", "dev", "--end", "2023-01-01"], input="\ny\n"
+        cli,
+        ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "dev", "--end", "2023-01-01"],
+        input="\ny\n",
     )
     assert_plan_success(result, "dev")
     assert "sqlmesh_example__dev.full_model: 2020-01-01 - 2023-01-01" in result.output
@@ -267,13 +295,35 @@ def test_plan_dev_create_from(runner, tmp_path):
     create_example_project(tmp_path)
 
     # create dev environment and backfill
-    runner.invoke(cli, ["--paths", tmp_path, "plan", "dev", "--no-prompts", "--auto-apply"])
+    runner.invoke(
+        cli,
+        [
+            "--log-file-dir",
+            tmp_path,
+            "--paths",
+            tmp_path,
+            "plan",
+            "dev",
+            "--no-prompts",
+            "--auto-apply",
+        ],
+    )
 
     # create dev2 environment from dev environment
     # Input: `y` to apply and virtual update
     result = runner.invoke(
         cli,
-        ["--paths", tmp_path, "plan", "dev2", "--create-from", "dev", "--include-unmodified"],
+        [
+            "--log-file-dir",
+            tmp_path,
+            "--paths",
+            tmp_path,
+            "plan",
+            "dev2",
+            "--create-from",
+            "dev",
+            "--include-unmodified",
+        ],
         input="y\n",
     )
     assert result.exit_code == 0
@@ -288,7 +338,9 @@ def test_plan_dev_no_prompts(runner, tmp_path):
 
     # plan for non-prod environment doesn't prompt to apply and doesn't
     # backfill if only `--no-prompts` is passed
-    result = runner.invoke(cli, ["--paths", tmp_path, "plan", "dev", "--no-prompts"])
+    result = runner.invoke(
+        cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "dev", "--no-prompts"]
+    )
     assert result.exit_code == 0
     assert "Apply - Backfill Tables [y/n]: " not in result.output
     assert "All model versions have been created successfully" not in result.output
@@ -300,7 +352,11 @@ def test_plan_dev_auto_apply(runner, tmp_path):
     create_example_project(tmp_path)
 
     # Input: enter for backfill start date prompt, enter for end date prompt
-    result = runner.invoke(cli, ["--paths", tmp_path, "plan", "dev", "--auto-apply"], input="\n\n")
+    result = runner.invoke(
+        cli,
+        ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "dev", "--auto-apply"],
+        input="\n\n",
+    )
     assert_plan_success(result, "dev")
 
 
@@ -309,7 +365,7 @@ def test_plan_dev_no_changes(runner, tmp_path):
     init_prod_and_backfill(runner, tmp_path)
 
     # Error if no changes made and `--include-unmodified` is not passed
-    result = runner.invoke(cli, ["--paths", tmp_path, "plan", "dev"])
+    result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "dev"])
     assert result.exit_code == 1
     assert (
         "Error: No changes were detected. Make a change or run with --include-unmodified"
@@ -319,7 +375,9 @@ def test_plan_dev_no_changes(runner, tmp_path):
     # No error if no changes made and `--include-unmodified` is passed
     # Input: `y` to apply and virtual update
     result = runner.invoke(
-        cli, ["--paths", tmp_path, "plan", "dev", "--include-unmodified"], input="y\n"
+        cli,
+        ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "dev", "--include-unmodified"],
+        input="y\n",
     )
     assert result.exit_code == 0
     assert_new_env(result, "dev")
@@ -334,7 +392,9 @@ def test_plan_nonbreaking(runner, tmp_path):
     update_incremental_model(tmp_path)
 
     # Input: `y` to apply and backfill
-    result = runner.invoke(cli, ["--paths", tmp_path, "plan"], input="y\n")
+    result = runner.invoke(
+        cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan"], input="y\n"
+    )
     assert result.exit_code == 0
     assert "Summary of differences against `prod`" in result.output
     assert "+  'a' AS new_col" in result.output
@@ -353,7 +413,9 @@ def test_plan_nonbreaking_noautocategorization(runner, tmp_path):
 
     # Input: `2` to classify change as non-breaking, `y` to apply and backfill
     result = runner.invoke(
-        cli, ["--paths", tmp_path, "plan", "--no-auto-categorization"], input="2\ny\n"
+        cli,
+        ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "--no-auto-categorization"],
+        input="2\ny\n",
     )
     assert result.exit_code == 0
     assert (
@@ -374,7 +436,9 @@ def test_plan_nonbreaking_nodiff(runner, tmp_path):
     update_incremental_model(tmp_path)
 
     # Input: `y` to apply and backfill
-    result = runner.invoke(cli, ["--paths", tmp_path, "plan", "--no-diff"], input="y\n")
+    result = runner.invoke(
+        cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "--no-diff"], input="y\n"
+    )
     assert result.exit_code == 0
     assert "+  'a' AS new_col" not in result.output
     assert_backfill_success(result)
@@ -388,7 +452,9 @@ def test_plan_breaking(runner, tmp_path):
 
     # full_model change makes test fail, so we pass `--skip-tests`
     # Input: `y` to apply and backfill
-    result = runner.invoke(cli, ["--paths", tmp_path, "plan", "--skip-tests"], input="y\n")
+    result = runner.invoke(
+        cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "--skip-tests"], input="y\n"
+    )
     assert result.exit_code == 0
     assert "+  item_id + 1 AS item_id," in result.output
     assert "Directly Modified: sqlmesh_example.full_model (Breaking)" in result.output
@@ -409,6 +475,8 @@ def test_plan_dev_select(runner, tmp_path):
     result = runner.invoke(
         cli,
         [
+            "--log-file-dir",
+            tmp_path,
             "--paths",
             tmp_path,
             "plan",
@@ -446,6 +514,8 @@ def test_plan_dev_backfill(runner, tmp_path):
     result = runner.invoke(
         cli,
         [
+            "--log-file-dir",
+            tmp_path,
             "--paths",
             tmp_path,
             "plan",
@@ -475,7 +545,7 @@ def test_run_no_prod(runner, tmp_path):
     create_example_project(tmp_path)
 
     # Error if no env specified and `prod` doesn't exist
-    result = runner.invoke(cli, ["--paths", tmp_path, "run"])
+    result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "run"])
     assert result.exit_code == 1
     assert "Error: Environment 'prod' was not found." in result.output
 
@@ -486,10 +556,14 @@ def test_run_dev(runner, tmp_path):
 
     # Create dev environment but DO NOT backfill
     # Input: `y` for virtual update
-    runner.invoke(cli, ["--paths", tmp_path, "plan", "dev", "--skip-backfill"], input="y\n")
+    runner.invoke(
+        cli,
+        ["--log-file-dir", tmp_path, "--paths", tmp_path, "plan", "dev", "--skip-backfill"],
+        input="y\n",
+    )
 
     # Confirm backfill occurs when we run non-backfilled dev env
-    result = runner.invoke(cli, ["--paths", tmp_path, "run", "dev"])
+    result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "run", "dev"])
     assert result.exit_code == 0
     assert_model_batches_executed(result)
 
@@ -500,7 +574,7 @@ def test_run_cron_not_elapsed(runner, tmp_path):
     init_prod_and_backfill(runner, tmp_path)
 
     # No error and no output if `prod` environment exists and cron has not elapsed
-    result = runner.invoke(cli, ["--paths", tmp_path, "run"])
+    result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "run"])
     assert result.exit_code == 0
     assert result.output == ""
 
@@ -514,7 +588,7 @@ def test_run_cron_elapsed(runner, tmp_path):
 
     # Run `prod` environment with daily cron elapsed
     with freeze_time("2023-01-02 00:01:00"):
-        result = runner.invoke(cli, ["--paths", tmp_path, "run"])
+        result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "run"])
 
     assert result.exit_code == 0
     assert_model_batches_executed(result)
@@ -531,7 +605,7 @@ def test_clean(runner, tmp_path):
     assert len(list(cache_path.iterdir())) > 0
 
     # Invoke the clean command
-    result = runner.invoke(cli, ["--paths", tmp_path, "clean"])
+    result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "clean"])
 
     # Confirm cache was cleared
     assert result.exit_code == 0
diff --git a/tests/core/engine_adapter/test_trino.py b/tests/core/engine_adapter/test_trino.py
index 51b4f16d..925c581f 100644
--- a/tests/core/engine_adapter/test_trino.py
+++ b/tests/core/engine_adapter/test_trino.py
@@ -5,7 +5,10 @@ import pytest
 from pytest_mock.plugin import MockerFixture
 from sqlglot import exp, parse_one
 
+import sqlmesh.core.dialect as d
 from sqlmesh.core.engine_adapter import TrinoEngineAdapter
+from sqlmesh.core.model import load_sql_based_model
+from sqlmesh.core.model.definition import SqlModel
 from tests.core.engine_adapter import to_sql_calls
 
 pytestmark = [pytest.mark.engine, pytest.mark.trino]
@@ -148,26 +151,36 @@ def test_partitioned_by_iceberg_transforms(
         return_value="datalake_iceberg",
     )
 
-    columns_to_types = {
-        "cola": exp.DataType.build("INT"),
-        "colb": exp.DataType.build("TEXT"),
-    }
+    expressions = d.parse(
+        f"""
+        MODEL (
+            name test_table,
+            partitioned_by (day(cola), truncate(colb, 8), colc),
+            kind INCREMENTAL_BY_TIME_RANGE(
+                time_column cola,
+            ),
+        );
+
+        SELECT 1::timestamp AS cola, 2::varchar as colb, 'foo' as colc;
+    """
+    )
+    model: SqlModel = t.cast(SqlModel, load_sql_based_model(expressions))
 
     adapter.create_table(
-        "test_table",
-        columns_to_types,
-        partitioned_by=[exp.to_column("day(cola)"), exp.to_column("truncate(colb, 8)")],
+        table_name=model.view_name,
+        columns_to_types=model.columns_to_types_or_raise,
+        partitioned_by=model.partitioned_by,
     )
 
     adapter.ctas(
-        "test_table",
-        parse_one("select 1"),  # type: ignore
-        partitioned_by=[exp.to_column("day(cola)"), exp.to_column("truncate(colb, 8)")],
+        table_name=model.view_name,
+        query_or_df=t.cast(exp.Query, model.query),
+        partitioned_by=model.partitioned_by,
     )
 
     assert to_sql_calls(adapter) == [
-        """CREATE TABLE IF NOT EXISTS "test_table" ("cola" INTEGER, "colb" VARCHAR) WITH (PARTITIONING=ARRAY['day(cola)', 'truncate(colb, 8)'])""",
-        """CREATE TABLE IF NOT EXISTS "test_table" WITH (PARTITIONING=ARRAY['day(cola)', 'truncate(colb, 8)']) AS SELECT 1""",
+        """CREATE TABLE IF NOT EXISTS "test_table" ("cola" TIMESTAMP, "colb" VARCHAR, "colc" VARCHAR) WITH (PARTITIONING=ARRAY['DAY("cola")', 'TRUNCATE("colb", 8)', '"colc"'])""",
+        """CREATE TABLE IF NOT EXISTS "test_table" WITH (PARTITIONING=ARRAY['DAY("cola")', 'TRUNCATE("colb", 8)', '"colc"']) AS SELECT CAST(1 AS TIMESTAMP) AS "cola", CAST(2 AS VARCHAR) AS "colb", \'foo\' AS "colc\"""",
     ]
 
 
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index ffa5eaaa..f5611c0a 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -245,17 +245,20 @@ def test_model_qualification():
 
 
 @pytest.mark.parametrize(
-    "partition_by_input, partition_by_output, expected_exception",
+    "partition_by_input, partition_by_output, output_dialect, expected_exception",
     [
-        ("a", ["`a`"], None),
-        ("(a, b)", ["`a`", "`b`"], None),
-        ("TIMESTAMP_TRUNC(`a`, DAY)", ["TIMESTAMP_TRUNC(`a`, DAY)"], None),
-        ("e", "", ConfigError),
-        ("(a, e)", "", ConfigError),
-        ("(a, a)", "", ConfigError),
+        ("a", ["`a`"], "bigquery", None),
+        ("(a, b)", ["`a`", "`b`"], "bigquery", None),
+        ("TIMESTAMP_TRUNC(`a`, DAY)", ["TIMESTAMP_TRUNC(`a`, DAY)"], "bigquery", None),
+        ("e", "", "bigquery", ConfigError),
+        ("(a, e)", "", "bigquery", ConfigError),
+        ("(a, a)", "", "bigquery", ConfigError),
+        ("(day(a),b)", ['DAY("a")', '"b"'], "trino", None),
     ],
 )
-def test_partitioned_by(partition_by_input, partition_by_output, expected_exception):
+def test_partitioned_by(
+    partition_by_input, partition_by_output, output_dialect, expected_exception
+):
     expressions = d.parse(
         f"""
         MODEL (
@@ -280,7 +283,9 @@ def test_partitioned_by(partition_by_input, partition_by_output, expected_except
             model.validate_definition()
     else:
         model.validate_definition()
-        assert [col.sql(dialect="bigquery") for col in model.partitioned_by] == partition_by_output
+        assert [
+            col.sql(dialect=output_dialect) for col in model.partitioned_by
+        ] == partition_by_output
 
 
 def test_no_model_statement():
@@ -2908,7 +2913,7 @@ def test_default_catalog_sql(assert_exp_eq):
     The system is not designed to actually support having an engine that doesn't support default catalog
     to start supporting it or the reverse of that. If that did happen then bugs would occur.
     """
-    HASH_WITH_CATALOG = "1074956187"
+    HASH_WITH_CATALOG = "3198762995"
 
     # Test setting default catalog doesn't change hash if it matches existing logic
     expressions = d.parse(
@@ -3239,3 +3244,43 @@ def test_depends_on_default_catalog_python():
 
     assert m.default_catalog == "catalog"
     assert m.depends_on == {'"catalog"."other"."table"'}
+
+
+def test_end_date():
+    expressions = d.parse(
+        f"""
+        MODEL (
+            name db.table,
+            kind INCREMENTAL_BY_TIME_RANGE (
+                time_column ts,
+            ),
+            start '2023-01-01',
+            end '2023-06-01'
+        );
+
+        SELECT 1::int AS a, 2::int AS b, now::timestamp as ts
+        """
+    )
+    model = load_sql_based_model(expressions)
+
+    assert model.start == "2023-01-01"
+    assert model.end == "2023-06-01"
+    assert model.interval_unit == IntervalUnit.DAY
+
+    with pytest.raises(ConfigError, match=".*Start date.+can't be greater than end date.*"):
+        load_sql_based_model(
+            d.parse(
+                f"""
+            MODEL (
+                name db.table,
+                kind INCREMENTAL_BY_TIME_RANGE (
+                    time_column ts,
+                ),
+                start '2024-01-01',
+                end '2023-06-01'
+            );
+
+            SELECT 1::int AS a, 2::int AS b, now::timestamp as ts
+            """
+            )
+        )
diff --git a/tests/core/test_scheduler.py b/tests/core/test_scheduler.py
index 5b548d49..40206455 100644
--- a/tests/core/test_scheduler.py
+++ b/tests/core/test_scheduler.py
@@ -10,6 +10,7 @@ from sqlmesh.core.model.kind import (
     IncrementalByUniqueKeyKind,
     TimeColumn,
 )
+from sqlmesh.core.node import IntervalUnit
 from sqlmesh.core.scheduler import Scheduler, compute_interval_params
 from sqlmesh.core.snapshot import Snapshot, SnapshotEvaluator
 from sqlmesh.utils.date import to_datetime
@@ -234,3 +235,41 @@ def test_circuit_breaker(scheduler: Scheduler):
             "2022-01-30",
             circuit_breaker=lambda: True,
         )
+
+
+def test_intervals_with_end_date_on_model(mocker: MockerFixture, make_snapshot):
+    snapshot: Snapshot = make_snapshot(
+        SqlModel(
+            name="name",
+            kind=IncrementalByTimeRangeKind(time_column="ds", batch_size=1),
+            interval_unit=IntervalUnit.DAY,
+            start="2023-01-01",
+            end="2023-01-31",
+            query=parse_one("SELECT ds FROM parent.tbl"),
+        )
+    )
+
+    snapshot_evaluator = SnapshotEvaluator(adapter=mocker.MagicMock(), ddl_concurrent_tasks=1)
+    scheduler = Scheduler(
+        snapshots=[snapshot],
+        snapshot_evaluator=snapshot_evaluator,
+        state_sync=mocker.MagicMock(),
+        max_workers=2,
+        default_catalog=None,
+    )
+
+    # generate for 1 year to show that the returned batches should only cover
+    # the range defined on the model itself
+    batches = scheduler.batches(start="2023-01-01", end="2024-01-01")[snapshot]
+
+    assert len(batches) == 31  # days in Jan 2023
+    assert batches[0] == (to_datetime("2023-01-01"), to_datetime("2023-01-02"))
+    assert batches[-1] == (to_datetime("2023-01-31"), to_datetime("2023-02-01"))
+
+    # generate for less than 1 month to ensure that the scheduler end date
+    # takes precedence over the model end date
+    batches = scheduler.batches(start="2023-01-01", end="2023-01-10")[snapshot]
+
+    assert len(batches) == 10
+    assert batches[0] == (to_datetime("2023-01-01"), to_datetime("2023-01-02"))
+    assert batches[-1] == (to_datetime("2023-01-10"), to_datetime("2023-01-11"))
diff --git a/tests/core/test_snapshot.py b/tests/core/test_snapshot.py
index 948a147e..caac7a3d 100644
--- a/tests/core/test_snapshot.py
+++ b/tests/core/test_snapshot.py
@@ -593,7 +593,7 @@ def test_fingerprint(model: Model, parent_model: Model):
     fingerprint = fingerprint_from_node(model, nodes={})
 
     original_fingerprint = SnapshotFingerprint(
-        data_hash="3648098745",
+        data_hash="372035512",
         metadata_hash="3575901791",
     )
 
@@ -692,7 +692,7 @@ def test_fingerprint_jinja_macros(model: Model):
         }
     )
     original_fingerprint = SnapshotFingerprint(
-        data_hash="3877498837",
+        data_hash="900027747",
         metadata_hash="3575901791",
     )
 
diff --git a/tests/integrations/jupyter/test_magics.py b/tests/integrations/jupyter/test_magics.py
index 32ba174d..88401375 100644
--- a/tests/integrations/jupyter/test_magics.py
+++ b/tests/integrations/jupyter/test_magics.py
@@ -46,9 +46,11 @@ def notebook(mocker: MockerFixture, ip):
 
 
 @pytest.fixture
-def sushi_context(copy_to_temp_path, notebook) -> Context:
+def sushi_context(copy_to_temp_path, notebook, tmp_path) -> Context:
     sushi_dir = copy_to_temp_path(SUSHI_EXAMPLE_PATH)[0]
-    notebook.run_line_magic(magic_name="context", line=str(sushi_dir))
+    notebook.run_line_magic(
+        magic_name="context", line=f"{str(sushi_dir)} --log-file-dir {tmp_path}"
+    )
     return notebook.user_ns["context"]
 
 
@@ -95,9 +97,11 @@ def get_all_html_output():
     return _convert
 
 
-def test_context(notebook, convert_all_html_output_to_text, get_all_html_output):
+def test_context(notebook, convert_all_html_output_to_text, get_all_html_output, tmp_path):
     with capture_output() as output:
-        notebook.run_line_magic(magic_name="context", line=str(SUSHI_EXAMPLE_PATH))
+        notebook.run_line_magic(
+            magic_name="context", line=f"{str(SUSHI_EXAMPLE_PATH)}  --log-file-dir {tmp_path}"
+        )
 
     assert output.stdout == ""
     assert output.stderr == ""