breaking: validate only correct model kinds support batch_size. (#665)

fix: full refresh should respect all intervals to get upstream backfills

Author: tobymao

examples/wursthall/models/db/item_d.sql

@@ -5,12 +5,11 @@ MODEL (
   cron '@daily',
   owner jen,
   start '2022-06-01 00:00:00+00:00',
-  batch_size 200
 );
 
 SELECT
   id AS item_id,
   item_name AS item_name,
   item_group AS item_group,
   item_price AS item_price
-FROM src.menu_item_details
+FROM src.menu_item_details

sqlmesh/core/model/__init__.py

@@ -19,3 +19,4 @@
     TimeColumn,
 )
 from sqlmesh.core.model.meta import ModelMeta
+from sqlmesh.core.model.seed import Seed

sqlmesh/core/model/kind.py

@@ -61,6 +61,14 @@ def is_seed(self) -> bool:
     def is_materialized(self) -> bool:
         return self.name not in (ModelKindName.VIEW, ModelKindName.EMBEDDED)
 
+    @property
+    def supports_batch_size(self) -> bool:
+        """Whether or not this model supports the batch_size property."""
+        return self.name in (
+            ModelKindName.INCREMENTAL_BY_TIME_RANGE,
+            ModelKindName.INCREMENTAL_BY_UNIQUE_KEY,
+        )
+
     @property
     def only_latest(self) -> bool:
         """Whether or not this model only cares about latest date to render."""

sqlmesh/core/model/meta.py

@@ -234,9 +234,12 @@ def _int_validator(cls, v: t.Any) -> t.Optional[int]:
     @root_validator
     def _kind_validator(cls, values: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]:
         kind = values.get("kind")
-        if kind and not kind.is_materialized:
-            if values.get("partitioned_by_"):
-                raise ValueError(f"partitioned_by field cannot be set for {kind} models")
+        if kind:
+            if not kind.is_materialized:
+                if values.get("partitioned_by_"):
+                    raise ValueError(f"partitioned_by field cannot be set for {kind} models")
+            if values.get("batch_size") and not kind.supports_batch_size:
+                raise ValueError(f"batch_size field cannot be set for {kind} models")
         return values
 
     @property

sqlmesh/core/snapshot/definition.py

@@ -484,19 +484,6 @@ def missing_intervals(
         if self.is_embedded_kind:
             return []
 
-        if self.is_full_kind or self.is_view_kind or self.is_seed_kind:
-            latest = latest or now()
-
-            latest_start, latest_end = self._inclusive_exclusive(
-                latest if is_date(latest) else self.model.cron_prev(self.model.cron_floor(latest)),
-                latest,
-            )
-            # if the latest ts is stored in the last interval, nothing is missing
-            # else returns the latest ts with the exclusive end ts.
-            if self.intervals and self.intervals[-1][1] >= latest_end:
-                return []
-            return [(latest_start, latest_end)]
-
         missing = []
         start_dt, end_dt = (to_datetime(ts) for ts in self._inclusive_exclusive(start, end))
         dates = tuple(croniter_range(start_dt, end_dt, self.model.normalized_cron()))

tests/core/test_integration.py

@@ -224,6 +224,7 @@ def change_model_kind(context: Context, kind: ModelKindName):
     if kind in (ModelKindName.VIEW, ModelKindName.EMBEDDED, ModelKindName.FULL):
         context.upsert_model(
             "sushi.items",
+            batch_size=None,
             partitioned_by=[],
             audits=[],
         )

tests/core/test_model.py

@@ -966,3 +966,25 @@ def test_star_expansion(assert_exp_eq) -> None:
         ) AS model2
         """,
     )
+
+
+def test_batch_size_validation():
+    expressions = parse(
+        """
+        MODEL (
+            name db.seed,
+            kind SEED (
+              path '../seeds/waiter_names.csv',
+              batch_size 100,
+            ),
+            columns (
+              id double,
+              alias varchar
+            ),
+            batch_size 100,
+        );
+    """
+    )
+
+    with pytest.raises(ConfigError) as ex:
+        load_model(expressions, path=Path("./examples/sushi/models/test_model.sql"))

tests/core/test_snapshot.py

@@ -10,9 +10,9 @@
 from sqlmesh.core.model import (
     IncrementalByTimeRangeKind,
     Model,
-    ModelKind,
-    ModelKindName,
+    Seed,
     SeedKind,
+    SeedModel,
     SqlModel,
     create_seed_model,
     load_model,
@@ -53,20 +53,6 @@ def model():
     )
 
 
-@pytest.fixture
-def full_refresh_model():
-    return SqlModel(
-        name="fr_model",
-        kind=ModelKind(name=ModelKindName.FULL),
-        owner="owner",
-        dialect="spark",
-        cron="1 0 * * *",
-        batch_size=30,
-        start="2020-01-01",
-        query=parse_one("SELECT @EACH([1, 2], x -> x), ds FROM parent.tbl"),
-    )
-
-
 @pytest.fixture
 def snapshot(
     model: Model,
@@ -86,25 +72,6 @@ def snapshot(
     return snapshot
 
 
-@pytest.fixture
-def full_refresh_snapshot(
-    full_refresh_model: Model,
-    parent_model: Model,
-    monkeypatch: MonkeyPatch,
-    mocker: MockerFixture,
-    make_snapshot,
-):
-    mock = mocker.Mock()
-    mock.return_value = to_datetime("2022-09-23T00:12:53+00:00")
-    monkeypatch.setattr("sqlmesh.utils.date.now", mock)
-    full_refresh_snapshot = make_snapshot(
-        full_refresh_model,
-        models={parent_model.name: parent_model, full_refresh_model.name: full_refresh_model},
-    )
-    full_refresh_snapshot.version = full_refresh_snapshot.fingerprint
-    return full_refresh_snapshot
-
-
 def test_json(snapshot: Snapshot):
     assert json.loads(snapshot.json()) == {
         "created_ts": 1663891973000,
@@ -242,31 +209,22 @@ def test_missing_intervals(snapshot: Snapshot):
     ]
 
 
-def test_missing_interval_latest(
-    full_refresh_snapshot: Snapshot, monkeypatch: MonkeyPatch, mocker: MockerFixture
-):
-    mock = mocker.Mock()
-    mock.return_value = to_datetime("2020-01-05T00:12:53+00:00")
-    monkeypatch.setattr("sqlmesh.core.snapshot.definition.now", mock)
-    full_refresh_snapshot.add_interval("2020-01-01", "2020-01-01")
-    assert full_refresh_snapshot.missing_intervals("2020-01-01", "2020-01-01", "2020-01-01") == []
-    assert full_refresh_snapshot.missing_intervals("2020-01-02", "2020-01-02", "2020-01-02") == [
-        (to_timestamp("2020-01-02"), to_timestamp("2020-01-03"))
-    ]
-    assert full_refresh_snapshot.missing_intervals("2020-01-02", "2020-01-03", "2020-01-03") == [
-        (to_timestamp("2020-01-03"), to_timestamp("2020-01-04"))
-    ]
-    assert full_refresh_snapshot.missing_intervals("2020-01-02", "2020-01-03", "2020-01-04") == [
-        (to_timestamp("2020-01-04"), to_timestamp("2020-01-05"))
+def test_seed_intervals(make_snapshot):
+    snapshot_a = make_snapshot(
+        SeedModel(
+            name="a",
+            kind=SeedKind(path="./path/to/seed"),
+            seed=Seed(content="content"),
+            depends_on=set(),
+        )
+    )
+
+    assert snapshot_a.missing_intervals("2020-01-01", "2020-01-01") == [
+        (to_timestamp("2020-01-01"), to_timestamp("2020-01-02"))
     ]
-    assert full_refresh_snapshot.missing_intervals(
-        "2020-01-02", "2020-01-03", "2020-01-03 01:00:00"
-    ) == [(to_timestamp("2020-01-02"), to_timestamp("2020-01-03"))]
-    assert full_refresh_snapshot.missing_intervals(
-        "2020-01-02", "2020-01-03", "2020-01-04 23:59:59"
-    ) == [(to_timestamp("2020-01-03"), to_timestamp("2020-01-04"))]
-    assert full_refresh_snapshot.missing_intervals("2020-01-02", "2020-01-03") == [
-        (to_timestamp("2020-01-04"), to_timestamp("2020-01-05"))
+    snapshot_a.add_interval("2020-01-01", "2020-01-01")
+    assert snapshot_a.missing_intervals("2020-01-02", "2020-01-02") == [
+        (to_timestamp("2020-01-02"), to_timestamp("2020-01-03"))
     ]
 
 

tests/dbt/test_config.py

@@ -112,7 +112,6 @@ def test_to_sqlmesh_fields(sushi_test_project: Project):
         cron="@hourly",
         meta={"stamp": "bar", "dialect": "duckdb"},
         owner="Sally",
-        batch_size=2,
     )
     context = DbtContext(project_name="Foo")
     context.target = DuckDbConfig(schema="foo")
@@ -128,7 +127,6 @@ def test_to_sqlmesh_fields(sushi_test_project: Project):
     assert model.stamp == "bar"
     assert model.dialect == "duckdb"
     assert model.owner == "Sally"
-    assert model.batch_size == 2
 
 
 def test_model_config_sql_no_config():