Repository: TobikoData/sqlmesh
Pull request #2287 (2024-03-16 02:56:02)

Feat: Add the ability to specify a model end date
This feature allows setting an explicit end date on an incremental model, rather than using the implicit end date of "right now".

The use-case is addressing historical / "archive" datasets that only have data for a certain date range (and typically these would be defined as forward-only as well).

In these cases, its there is no use in generating new intervals after a certain date.

tobymao: this looks good to me, thanks so much for the pr @erindru 
izeigerman: @erindru I've just realized that one thing to keep in mind, that this doesn't prevent model DAGs on Airflow from continuing the evaluation because they don't rely on missing intervals but on whatever intervals Airflow itself supplies. 

We should make sure to set the `end_date` here as well: https://github.com/TobikoData/sqlmesh/blob/main/sqlmesh/schedulers/airflow/dag_generator.py#L114

Patch:
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/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/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/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/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/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/tests/core/test_model.py b/tests/core/test_model.py
index 9a3dd030..f5611c0a 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -3244,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"))