Fix: complete lookback support for custom materializations

sqlmesh/core/model/meta.py

@@ -29,7 +29,6 @@
     SCDType2ByTimeKind,
     TimeColumn,
     ViewKind,
-    _IncrementalBy,
     model_kind_validator,
     OnAdditiveChange,
 )
@@ -414,7 +413,7 @@ def column_descriptions(self) -> t.Dict[str, str]:
     @property
     def lookback(self) -> int:
         """The incremental lookback window."""
-        return (self.kind.lookback if isinstance(self.kind, _IncrementalBy) else 0) or 0
+        return getattr(self.kind, "lookback", 0) or 0
 
     def lookback_start(self, start: TimeLike) -> TimeLike:
         if self.lookback == 0:

tests/core/test_model.py

@@ -7685,6 +7685,75 @@ class MyTestStrategy(CustomMaterialization):
     )
 
 
+def test_custom_kind_lookback_property():
+    """Test that CustomKind's lookback property is correctly accessed via ModelMeta.lookback.
+
+    This test verifies the fix for issue #5268 where CustomKind models were not respecting
+    the lookback parameter because the isinstance check for _IncrementalBy failed.
+    """
+
+    # Test 1: CustomKind with lookback = 3
+    class MyTestStrategy(CustomMaterialization):
+        pass
+
+    expressions = d.parse(
+        """
+        MODEL (
+            name db.custom_table,
+            kind CUSTOM (
+                materialization 'MyTestStrategy',
+                lookback 3
+            )
+        );
+        SELECT a, b FROM upstream
+        """
+    )
+
+    model = load_sql_based_model(expressions)
+    assert model.kind.is_custom
+
+    # Verify that the kind itself has lookback = 3
+    kind = t.cast(CustomKind, model.kind)
+    assert kind.lookback == 3
+
+    # The bug: model.lookback should return 3, but with the old implementation
+    # using isinstance(self.kind, _IncrementalBy), it would return 0
+    assert model.lookback == 3, "CustomKind lookback not accessible via model.lookback property"
+
+    # Test 2: CustomKind without lookback (should default to 0)
+    expressions_no_lookback = d.parse(
+        """
+        MODEL (
+            name db.custom_table_no_lookback,
+            kind CUSTOM (
+                materialization 'MyTestStrategy'
+            )
+        );
+        SELECT a, b FROM upstream
+        """
+    )
+
+    model_no_lookback = load_sql_based_model(expressions_no_lookback)
+    assert model_no_lookback.lookback == 0
+
+    # Test 3: Ensure IncrementalByTimeRangeKind still works correctly
+    incremental_expressions = d.parse(
+        """
+        MODEL (
+            name db.incremental_table,
+            kind INCREMENTAL_BY_TIME_RANGE (
+                time_column ds,
+                lookback 5
+            )
+        );
+        SELECT ds, a, b FROM upstream
+        """
+    )
+
+    incremental_model = load_sql_based_model(incremental_expressions)
+    assert incremental_model.lookback == 5
+
+
 def test_time_column_format_in_custom_kind():
     class TimeColumnCustomKind(CustomKind):  # type: ignore[no-untyped-def]
         _time_column: TimeColumn

tests/core/test_snapshot.py

@@ -679,6 +679,43 @@ def test_lookback(make_snapshot):
     assert snapshot.missing_intervals("2023-01-28", "2023-01-30", "2023-01-31 04:00:00") == []
 
 
+def test_lookback_custom_materialization(make_snapshot):
+    from sqlmesh import CustomMaterialization
+
+    class MyTestStrategy(CustomMaterialization):
+        pass
+
+    expressions = parse(
+        """
+        MODEL (
+            name name,
+            kind CUSTOM (
+                materialization 'MyTestStrategy',
+                lookback 2
+            ),
+            start '2023-01-01',
+            cron '0 5 * * *',
+        );
+
+        SELECT ds FROM parent.tbl
+        """
+    )
+
+    snapshot = make_snapshot(load_sql_based_model(expressions))
+
+    assert snapshot.missing_intervals("2023-01-01", "2023-01-01") == [
+        (to_timestamp("2023-01-01"), to_timestamp("2023-01-02")),
+    ]
+
+    snapshot.add_interval("2023-01-01", "2023-01-04")
+    assert snapshot.missing_intervals("2023-01-01", "2023-01-04") == []
+    assert snapshot.missing_intervals("2023-01-01", "2023-01-05") == [
+        (to_timestamp("2023-01-03"), to_timestamp("2023-01-04")),
+        (to_timestamp("2023-01-04"), to_timestamp("2023-01-05")),
+        (to_timestamp("2023-01-05"), to_timestamp("2023-01-06")),
+    ]
+
+
 def test_seed_intervals(make_snapshot):
     snapshot_a = make_snapshot(
         SeedModel(