Fix: Warn when SQLMesh automatically adjusts a restatement range to c…

…over the whole model

sqlmesh/core/snapshot/definition.py

@@ -715,9 +715,29 @@ def get_removal_interval(
                 a run.
         """
         end = execution_time or now_timestamp() if self.depends_on_past else end
+        removal_interval = self.inclusive_exclusive(start, end, strict)
+
         if not is_preview and self.full_history_restatement_only and self.intervals:
-            start = self.intervals[0][0]
-        return self.inclusive_exclusive(start, end, strict)
+            expanded_removal_interval = self.inclusive_exclusive(self.intervals[0][0], end, strict)
+            requested_start, requested_end = removal_interval
+            expanded_start, expanded_end = expanded_removal_interval
+
+            is_subset = requested_start > expanded_start or requested_end < expanded_end
+
+            # only warn if the requested removal interval was a subset of the actual model intervals and was automatically expanded
+            # if the requested interval was the same or wider than the actual model intervals, no need to warn
+            if removal_interval != expanded_removal_interval and is_subset:
+                from sqlmesh.core.console import get_console
+
+                get_console().log_warning(
+                    f"Model '{self.model.name}' is '{self.model_kind_name}' which does not support partial restatement.\n"
+                    f"Expanding the requested restatement intervals from [{to_ts(requested_start)} - {to_ts(requested_end)}] "
+                    f"to [{to_ts(expanded_start)} - {to_ts(expanded_end)}] in order to fully restate the model."
+                )
+
+            removal_interval = expanded_removal_interval
+
+        return removal_interval
 
     def inclusive_exclusive(
         self,

tests/core/test_snapshot.py

@@ -23,6 +23,7 @@
 from sqlmesh.core.model import (
     FullKind,
     IncrementalByTimeRangeKind,
+    IncrementalByUniqueKeyKind,
     IncrementalUnmanagedKind,
     Model,
     Seed,
@@ -61,6 +62,7 @@
 from sqlmesh.utils.date import DatetimeRanges, to_date, to_datetime, to_timestamp
 from sqlmesh.utils.errors import SQLMeshError
 from sqlmesh.utils.jinja import JinjaMacroRegistry, MacroInfo
+from sqlmesh.core.console import get_console
 
 
 @pytest.fixture
@@ -752,6 +754,63 @@ def test_get_removal_intervals_full_history_restatement_model(make_snapshot):
     assert interval == (to_timestamp("2023-01-01"), execution_time)
 
 
+def test_get_removal_intervals_warns_when_requested_range_automatically_widened(
+    make_snapshot: t.Callable[..., Snapshot], mocker: MockerFixture
+):
+    mock_logger = mocker.patch.object(get_console(), "log_warning")
+
+    # INCREMENTAL_BY_UNIQUE_KEY should warn
+    snapshot = make_snapshot(
+        SqlModel(
+            name="name",
+            kind=IncrementalByUniqueKeyKind(unique_key=[exp.to_column("id")]),
+            query=parse_one("select id from src"),
+        )
+    )
+
+    assert not snapshot.intervals
+    assert snapshot.full_history_restatement_only
+
+    snapshot.add_interval("2020-01-01", "2020-01-10")
+
+    # should warn if requested intervals are a subset of actual intervals and thus are automatically expanded
+    snapshot.get_removal_interval("2020-01-05", "2020-01-06")
+
+    msg = mock_logger.call_args[0][0]
+    assert "does not support partial restatement" in msg
+    assert "Expanding the requested restatement intervals" in msg
+
+    # should not warn if requested intervals are equal to actual intervals
+    mock_logger.reset_mock()
+
+    snapshot.get_removal_interval("2020-01-01", "2020-01-10")
+    mock_logger.assert_not_called()
+
+    # should not warn if requested intervals are a superset of actual intervals
+    mock_logger.reset_mock()
+
+    snapshot.get_removal_interval("2019-12-30", "2020-01-15")
+    mock_logger.assert_not_called()
+
+    # should not warn on models that support partial restatement, such as INCREMENTAL_BY_TIME_RANGE
+    mock_logger.reset_mock()
+    snapshot = make_snapshot(
+        SqlModel(
+            name="name",
+            kind=IncrementalByTimeRangeKind(time_column=TimeColumn(column="ds")),
+            query=parse_one("select ds from src"),
+        )
+    )
+
+    assert not snapshot.intervals
+    assert not snapshot.full_history_restatement_only
+
+    snapshot.add_interval("2020-01-01", "2020-01-10")
+
+    snapshot.get_removal_interval("2020-01-05", "2020-01-06")
+    mock_logger.assert_not_called()
+
+
 each_macro = lambda: "test"  # noqa: E731