Feat!: Add depends_on_past through self-reference check support (Built-in only) (#917)

* add full with history

* add full with history model with test

* fmt model

* cleanup

* adjust fingerprint

* fix integration tests

* remove full with history kind

* feedback

* fix after rebase

* update self reference name

* fix name

* remove properties for getting references

* add back extra line

* cleanup

* add migration script

* rename to depends_on_past

Author: eakmanrq

examples/sushi/models/customer_revenue_lifetime.sql

@@ -0,0 +1,58 @@
+/*
+    Table of lifetime customer revenue.
+    Date is available to get lifetime value up to a certain date.
+    Use latest date to get current lifetime value.
+*/
+MODEL (
+  name sushi.customer_revenue_lifetime,
+  kind incremental_by_time_range (
+    time_column (ds, 'YYYY-MM-dd'),
+    batch_size 1
+  ),
+  owner jen,
+  cron '@daily',
+  dialect hive,
+  columns (
+    customer_id INT,
+    revenue DOUBLE,
+    ds STRING
+  )
+);
+
+WITH order_total AS (
+  SELECT
+    oi.order_id AS order_id,
+    SUM(oi.quantity * i.price) AS total
+  FROM sushi.order_items AS oi
+  LEFT JOIN sushi.items AS i
+    ON oi.item_id = i.id AND oi.ds = i.ds
+  WHERE
+    oi.ds = @end_ds
+  GROUP BY
+    oi.order_id
+), incremental_total AS (
+  SELECT
+    o.customer_id::INT AS customer_id,
+    SUM(ot.total)::DOUBLE AS revenue,
+  FROM sushi.orders AS o
+  LEFT JOIN order_total AS ot
+    ON o.id = ot.order_id
+  WHERE
+    o.ds = @end_ds
+  GROUP BY
+    o.customer_id
+), prev_total AS (
+  SELECT
+    crl.customer_id,
+    crl.revenue
+  FROM sushi.customer_revenue_lifetime AS crl
+  WHERE
+    crl.ds = DATE_FORMAT(@end_date - INTERVAL 1 DAY, 'YYYY-MM-dd')
+)
+SELECT
+  COALESCE(it.customer_id, prev_total.customer_id) AS customer_id, /* Customer id */
+  COALESCE(it.revenue, 0) + COALESCE(prev_total.revenue, 0) AS revenue, /* Lifetime revenue from this customer */
+  @end_ds AS ds /* End date of the lifetime calculation */
+FROM incremental_total AS it
+FULL OUTER JOIN prev_total AS prev_total
+  ON it.customer_id = prev_total.customer_id

sqlmesh/core/config/model.py

@@ -3,7 +3,7 @@
 import typing as t
 
 from sqlmesh.core.config.base import BaseConfig
-from sqlmesh.core.model.kind import ModelKind, model_kind_validator
+from sqlmesh.core.model.kind import ModelKind
 from sqlmesh.utils.date import TimeLike
 
 
@@ -34,4 +34,4 @@ class ModelDefaultsConfig(BaseConfig):
     batch_size: t.Optional[int]
     storage_format: t.Optional[str]
 
-    _model_kind_validator = model_kind_validator
+    _model_kind_validator = ModelKind.field_validator()

sqlmesh/core/model/definition.py

@@ -104,6 +104,7 @@ class _Model(ModelMeta, frozen=True):
 
     _path: Path = Path()
     _depends_on: t.Optional[t.Set[str]] = None
+    _depends_on_past: t.Optional[bool] = None
     _column_descriptions: t.Optional[t.Dict[str, str]] = None
 
     _expressions_validator = expression_validator
@@ -449,6 +450,15 @@ def is_python(self) -> bool:
     def is_seed(self) -> bool:
         return False
 
+    @property
+    def depends_on_past(self) -> bool:
+        if self._depends_on_past is None:
+            self._depends_on_past = (
+                self.kind.is_incremental_by_unique_key
+                or self.name in _find_tables([self.render_query()])
+            )
+        return self._depends_on_past
+
     def validate_definition(self) -> None:
         """Validates the model's definition.
 
@@ -817,6 +827,10 @@ def seed_path(self) -> Path:
             return self._path.parent / seed_path
         return seed_path
 
+    @property
+    def depends_on_past(self) -> bool:
+        return False
+
     def to_dehydrated(self) -> SeedModel:
         """Creates a dehydrated copy of this model.
 

sqlmesh/core/model/kind.py

@@ -59,7 +59,7 @@ def is_materialized(self) -> bool:
     @property
     def only_latest(self) -> bool:
         """Whether or not this model only cares about latest date to render."""
-        return self.model_kind_name in (ModelKindName.VIEW, ModelKindName.FULL)
+        return self.is_view or self.is_full
 
 
 class ModelKindName(str, ModelKindMixin, Enum):
@@ -81,18 +81,80 @@ def model_kind_name(self) -> ModelKindName:
 class ModelKind(PydanticModel, ModelKindMixin):
     name: ModelKindName
 
-    def to_expression(self, **kwargs: t.Any) -> d.ModelKind:
-        return d.ModelKind(this=self.name.value.upper(), **kwargs)
+    @classmethod
+    def field_validator(cls) -> classmethod:
+        def _model_kind_validator(v: t.Any) -> ModelKind:
+            if isinstance(v, ModelKind):
+                return v
+
+            if isinstance(v, d.ModelKind):
+                name = v.this
+                props = {prop.name: prop.args.get("value") for prop in v.expressions}
+                klass: t.Type[ModelKind] = ModelKind
+                if name == ModelKindName.INCREMENTAL_BY_TIME_RANGE:
+                    klass = IncrementalByTimeRangeKind
+                elif name == ModelKindName.INCREMENTAL_BY_UNIQUE_KEY:
+                    klass = IncrementalByUniqueKeyKind
+                elif name == ModelKindName.SEED:
+                    klass = SeedKind
+                else:
+                    props["name"] = ModelKindName(name)
+                return klass(**props)
+
+            if isinstance(v, dict):
+                if v.get("name") == ModelKindName.INCREMENTAL_BY_TIME_RANGE:
+                    klass = IncrementalByTimeRangeKind
+                elif v.get("name") == ModelKindName.INCREMENTAL_BY_UNIQUE_KEY:
+                    klass = IncrementalByUniqueKeyKind
+                elif v.get("name") == ModelKindName.SEED:
+                    klass = SeedKind
+                else:
+                    klass = ModelKind
+                return klass(**v)
+
+            name = (v.name if isinstance(v, exp.Expression) else str(v)).upper()
+
+            try:
+                return ModelKind(name=ModelKindName(name))
+            except ValueError:
+                raise ConfigError(f"Invalid model kind '{name}'")
+
+        return validator("kind", pre=True, allow_reuse=True)(_model_kind_validator)
 
     @property
     def model_kind_name(self) -> ModelKindName:
         return self.name
 
+    def to_expression(self, **kwargs: t.Any) -> d.ModelKind:
+        return d.ModelKind(this=self.name.value.upper(), **kwargs)
+
 
 class TimeColumn(PydanticModel):
     column: str
     format: t.Optional[str] = None
 
+    @classmethod
+    def field_validator(cls) -> classmethod:
+        def _time_column_validator(v: t.Any) -> TimeColumn:
+            if isinstance(v, exp.Tuple):
+                kwargs = {
+                    key: v.expressions[i].name
+                    for i, key in enumerate(("column", "format")[: len(v.expressions)])
+                }
+                return TimeColumn(**kwargs)
+
+            if isinstance(v, exp.Identifier):
+                return TimeColumn(column=v.name)
+
+            if isinstance(v, exp.Expression):
+                return TimeColumn(column=v.name)
+
+            if isinstance(v, str):
+                return TimeColumn(column=v)
+            return v
+
+        return validator("time_column", pre=True, allow_reuse=True)(_time_column_validator)
+
     @validator("column", pre=True)
     def _column_validator(cls, v: str) -> str:
         if not v:
@@ -126,6 +188,9 @@ def to_expression(self, dialect: str) -> exp.Column | exp.Tuple:
             ]
         )
 
+    def to_property(self, dialect: str = "") -> exp.Property:
+        return exp.Property(this="time_column", value=self.to_expression(dialect))
+
 
 class _Incremental(ModelKind):
     batch_size: t.Optional[int]
@@ -156,31 +221,10 @@ class IncrementalByTimeRangeKind(_Incremental):
     name: ModelKindName = Field(ModelKindName.INCREMENTAL_BY_TIME_RANGE, const=True)
     time_column: TimeColumn
 
-    @validator("time_column", pre=True)
-    def _parse_time_column(cls, v: t.Any) -> TimeColumn:
-        if isinstance(v, exp.Tuple):
-            kwargs = {
-                key: v.expressions[i].name
-                for i, key in enumerate(("column", "format")[: len(v.expressions)])
-            }
-            return TimeColumn(**kwargs)
-
-        if isinstance(v, exp.Identifier):
-            return TimeColumn(column=v.name)
-
-        if isinstance(v, exp.Expression):
-            return TimeColumn(column=v.name)
-
-        if isinstance(v, str):
-            return TimeColumn(column=v)
-        return v
+    _time_column_validator = TimeColumn.field_validator()
 
     def to_expression(self, dialect: str = "", **kwargs: t.Any) -> d.ModelKind:
-        return super().to_expression(
-            expressions=[
-                exp.Property(this="time_column", value=self.time_column.to_expression(dialect))
-            ],
-        )
+        return super().to_expression(expressions=[self.time_column.to_property(dialect)])
 
 
 class IncrementalByUniqueKeyKind(_Incremental):
@@ -228,43 +272,3 @@ def to_expression(self, **kwargs: t.Any) -> d.ModelKind:
                 ),
             ],
         )
-
-
-def _model_kind_validator(v: t.Any) -> ModelKind:
-    if isinstance(v, ModelKind):
-        return v
-
-    if isinstance(v, d.ModelKind):
-        name = v.this
-        props = {prop.name: prop.args.get("value") for prop in v.expressions}
-        klass: t.Type[ModelKind] = ModelKind
-        if name == ModelKindName.INCREMENTAL_BY_TIME_RANGE:
-            klass = IncrementalByTimeRangeKind
-        elif name == ModelKindName.INCREMENTAL_BY_UNIQUE_KEY:
-            klass = IncrementalByUniqueKeyKind
-        elif name == ModelKindName.SEED:
-            klass = SeedKind
-        else:
-            props["name"] = ModelKindName(name)
-        return klass(**props)
-
-    if isinstance(v, dict):
-        if v.get("name") == ModelKindName.INCREMENTAL_BY_TIME_RANGE:
-            klass = IncrementalByTimeRangeKind
-        elif v.get("name") == ModelKindName.INCREMENTAL_BY_UNIQUE_KEY:
-            klass = IncrementalByUniqueKeyKind
-        elif v.get("name") == ModelKindName.SEED:
-            klass = SeedKind
-        else:
-            klass = ModelKind
-        return klass(**v)
-
-    name = (v.name if isinstance(v, exp.Expression) else str(v)).upper()
-
-    try:
-        return ModelKind(name=ModelKindName(name))
-    except ValueError:
-        raise ConfigError(f"Invalid model kind '{name}'")
-
-
-model_kind_validator = validator("kind", pre=True, allow_reuse=True)(_model_kind_validator)

sqlmesh/core/model/meta.py

@@ -10,13 +10,11 @@
 
 from sqlmesh.core import dialect as d
 from sqlmesh.core.model.kind import (
-    IncrementalByTimeRangeKind,
     IncrementalByUniqueKeyKind,
     ModelKind,
     ModelKindName,
     TimeColumn,
     _Incremental,
-    model_kind_validator,
 )
 from sqlmesh.utils import unique
 from sqlmesh.utils.date import TimeLike, preserve_time_like_kind, to_datetime
@@ -64,7 +62,7 @@ class ModelMeta(PydanticModel):
     _croniter: t.Optional[croniter] = None
     _interval_unit: t.Optional[IntervalUnit] = None
 
-    _model_kind_validator = model_kind_validator
+    _model_kind_validator = ModelKind.field_validator()
 
     @validator("audits", pre=True)
     def _audits_validator(cls, v: t.Any) -> t.Any:
@@ -174,9 +172,8 @@ def _kind_validator(cls, values: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]:
 
     @property
     def time_column(self) -> t.Optional[TimeColumn]:
-        if isinstance(self.kind, IncrementalByTimeRangeKind):
-            return self.kind.time_column
-        return None
+        """The time column for incremental models."""
+        return getattr(self.kind, "time_column", None)
 
     @property
     def unique_key(self) -> t.List[str]:
@@ -215,7 +212,7 @@ def lookback_delta(self) -> timedelta:
     @property
     def batch_size(self) -> t.Optional[int]:
         """The maximal number of units in a single task for a backfill."""
-        return self.kind.batch_size if isinstance(self.kind, _Incremental) else None
+        return getattr(self.kind, "batch_size", None)
 
     def interval_unit(self, sample_size: int = 10) -> IntervalUnit:
         """Returns the IntervalUnit of the model

sqlmesh/core/scheduler.py

@@ -272,12 +272,11 @@ def _dag(self, batches: SnapshotToBatches) -> DAG[SchedulingUnit]:
             ]
             for i, interval in enumerate(intervals):
                 dag.add((snapshot, interval), upstream_dependencies)
-                if snapshot.is_incremental_by_unique_key:
+                if snapshot.depends_on_past:
                     dag.add(
                         (snapshot, interval),
                         [(snapshot, _interval) for _interval in intervals[:i]],
                     )
-
         return dag
 
 

sqlmesh/core/snapshot/definition.py

@@ -803,6 +803,11 @@ def start(self) -> t.Optional[datetime]:
     def model_kind_name(self) -> ModelKindName:
         return self.model.kind.name
 
+    @property
+    def depends_on_past(self) -> bool:
+        """Whether or not this models depends on past intervals to be accurate before loading following intervals."""
+        return self.model.depends_on_past
+
     def _ensure_categorized(self) -> None:
         if not self.change_category:
             raise SQLMeshError(f"Snapshot {self.snapshot_id} has not been categorized yet.")

sqlmesh/core/snapshot/evaluator.py

@@ -117,7 +117,7 @@ def apply(query_or_df: QueryOrDF, index: int = 0) -> None:
                 logger.info("Inserting batch (%s, %s) into %s'", start, end, table_name)
                 if snapshot.is_incremental_by_time_range:
                     # A model's time_column could be None but
-                    # it shouldn't be for an incremental by time range model
+                    # it shouldn't be for time based loads
                     assert model.time_column
                     self.adapter.insert_overwrite_by_time_partition(
                         table_name,

sqlmesh/migrations/v0010_seed_hash_batch_size.py

@@ -0,0 +1,5 @@
+"""Seed metadata hashes now correctly include the batch_size."""
+
+
+def migrate(state_sync):  # type: ignore
+    pass