Feat: Introduce an additional 'incremental_unmanaged' model kind to support existing incremental mode in dbt projects (#1173)

Author: izeigerman

docs/integrations/dbt.md

@@ -77,14 +77,16 @@ To enable incremental_by_time_range incrementality, the model configuration shou
 
 ### Incremental logic
 
-SQLMesh will ignore dbt's incremental jinja block `{% if is_incremental() %}` and requires a new jinja block gated by `{% if sqlmesh_incremental is defined %}`. The new block should contain the `WHERE` clause selecting the time interval. 
+SQLMesh requires a new jinja block gated by `{% if sqlmesh_incremental is defined %}`. The new block should supersede the existing `{% if is_incremental() %}` block and contain the `WHERE` clause selecting the time interval.
 
 For example, the SQL `WHERE` clause with the "ds" column goes in a new jinja block gated by `{% if sqlmesh_incremental is defined %}` as follows:
 
 ```bash
 > {% if sqlmesh_incremental is defined %}
 >   WHERE
 >     ds BETWEEN '{{ start_ds }}' AND '{{ end_ds }}'
+> {% elif is_incremental() %}
+>   ; < your existing is_incremental block >
 > {% endif %}
 ```
 

sqlmesh/core/engine_adapter/base.py

@@ -592,6 +592,16 @@ def _insert_append_pandas_df(
                 ):
                     self.execute(exp.insert(expression, table_name, columns=column_names))
 
+    def insert_overwrite(
+        self,
+        table_name: TableName,
+        query_or_df: QueryOrDF,
+        columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
+    ) -> None:
+        self._insert_overwrite_by_condition(
+            table_name, query_or_df, columns_to_types=columns_to_types
+        )
+
     def insert_overwrite_by_time_partition(
         self,
         table_name: TableName,

sqlmesh/core/model/__init__.py

@@ -14,6 +14,7 @@
 from sqlmesh.core.model.kind import (
     IncrementalByTimeRangeKind,
     IncrementalByUniqueKeyKind,
+    IncrementalUnmanagedKind,
     ModelKind,
     ModelKindMixin,
     ModelKindName,

sqlmesh/core/model/common.py

@@ -67,6 +67,7 @@ def parse_bool(v: t.Any) -> bool:
     "blocking",
     "forward_only",
     "disable_restatement",
+    "insert_overwrite",
     pre=True,
     allow_reuse=True,
     check_fields=False,

sqlmesh/core/model/definition.py

@@ -29,7 +29,6 @@
     IncrementalByUniqueKeyKind,
     ModelKindName,
     SeedKind,
-    _Incremental,
 )
 from sqlmesh.core.model.meta import ModelMeta
 from sqlmesh.core.model.seed import Seed, create_seed
@@ -574,11 +573,11 @@ def depends_on_past(self) -> bool:
 
     @property
     def forward_only(self) -> bool:
-        return isinstance(self.kind, _Incremental) and self.kind.forward_only
+        return getattr(self.kind, "forward_only", False)
 
     @property
     def disable_restatement(self) -> bool:
-        return isinstance(self.kind, _Incremental) and self.kind.disable_restatement
+        return getattr(self.kind, "disable_restatement", False)
 
     def validate_definition(self) -> None:
         """Validates the model's definition.
@@ -620,7 +619,17 @@ def validate_definition(self) -> None:
 
         if self.kind.is_incremental_by_time_range and not self.time_column:
             raise_config_error(
-                "Incremental by time range models must have a time_column field.",
+                "Incremental by time range models must have a time_column field",
+                self._path,
+            )
+
+        if (
+            self.kind.is_incremental_unmanaged
+            and getattr(self.kind, "insert_overwrite", False)
+            and not self.partitioned_by_
+        ):
+            raise_config_error(
+                "Unmanaged incremental models with insert / overwrite enabled must specify the partitioned_by field",
                 self._path,
             )
 

sqlmesh/core/model/kind.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import sys
 import typing as t
 from enum import Enum
 
@@ -13,6 +14,11 @@
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.pydantic import PydanticModel
 
+if sys.version_info >= (3, 9):
+    from typing import Literal
+else:
+    from typing_extensions import Literal
+
 
 class ModelKindMixin:
     @property
@@ -28,6 +34,10 @@ def is_incremental_by_time_range(self) -> bool:
     def is_incremental_by_unique_key(self) -> bool:
         return self.model_kind_name == ModelKindName.INCREMENTAL_BY_UNIQUE_KEY
 
+    @property
+    def is_incremental_unmanaged(self) -> bool:
+        return self.model_kind_name == ModelKindName.INCREMENTAL_UNMANAGED
+
     @property
     def is_full(self) -> bool:
         return self.model_kind_name == ModelKindName.FULL
@@ -60,14 +70,15 @@ 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.is_view or self.is_full
+        return self.is_view or self.is_full or self.is_incremental_unmanaged
 
 
 class ModelKindName(str, ModelKindMixin, Enum):
     """The kind of model, determining how this data is computed and stored in the warehouse."""
 
     INCREMENTAL_BY_TIME_RANGE = "INCREMENTAL_BY_TIME_RANGE"
     INCREMENTAL_BY_UNIQUE_KEY = "INCREMENTAL_BY_UNIQUE_KEY"
+    INCREMENTAL_UNMANAGED = "INCREMENTAL_UNMANAGED"
     FULL = "FULL"
     VIEW = "VIEW"
     EMBEDDED = "EMBEDDED"
@@ -246,6 +257,14 @@ def _parse_unique_key(cls, v: t.Any) -> t.List[str]:
         return [i.this if isinstance(i, exp.Identifier) else str(i) for i in v]
 
 
+class IncrementalUnmanagedKind(ModelKind):
+    name: ModelKindName = Field(ModelKindName.INCREMENTAL_UNMANAGED, const=True)
+    insert_overwrite: bool = False
+    disable_restatement: Literal[True] = True
+
+    _bool_validator = bool_validator
+
+
 class ViewKind(ModelKind):
     name: ModelKindName = Field(ModelKindName.VIEW, const=True)
     materialized: bool = False

sqlmesh/core/plan/definition.py

@@ -441,13 +441,18 @@ def _add_restatements(self, restate_models: t.Iterable[str]) -> None:
 
             snapshots = self.context_diff.snapshots
             downstream = [
-                d
-                for d in downstream
-                if snapshots[d].is_materialized
-                and (not snapshots[d].model.disable_restatement or self.is_dev)
-                and not snapshots[d].is_seed
+                d for d in downstream if snapshots[d].is_materialized and not snapshots[d].is_seed
             ]
 
+            if not self.is_dev:
+                models_with_disabled_restatement = [
+                    f"'{d}'" for d in downstream if snapshots[d].model.disable_restatement
+                ]
+                if models_with_disabled_restatement:
+                    raise PlanError(
+                        f"Restatement is disabled for models: {', '.join(models_with_disabled_restatement)}."
+                    )
+
             if not downstream:
                 raise PlanError(
                     f"Cannot restate from '{table}'. Either such model doesn't exist, no other materialized model references it, or restatement was disabled fror this model."

sqlmesh/core/snapshot/evaluator.py

@@ -34,7 +34,7 @@
 from sqlmesh.core.audit import BUILT_IN_AUDITS, AuditResult
 from sqlmesh.core.engine_adapter import EngineAdapter, TransactionType
 from sqlmesh.core.engine_adapter.base import InsertOverwriteStrategy
-from sqlmesh.core.model import Model, ViewKind
+from sqlmesh.core.model import IncrementalUnmanagedKind, Model, ViewKind
 from sqlmesh.core.snapshot import (
     QualifiedViewName,
     Snapshot,
@@ -130,6 +130,7 @@ def apply(query_or_df: QueryOrDF, index: int = 0) -> None:
             start=start,
             end=end,
             latest=latest,
+            has_intervals=bool(snapshot.intervals),
             **kwargs,
         )
 
@@ -488,6 +489,8 @@ def _evaluation_strategy(snapshot: SnapshotInfoLike, adapter: EngineAdapter) ->
         klass = IncrementalByTimeRangeStrategy
     elif snapshot.is_incremental_by_unique_key:
         klass = IncrementalByUniqueKeyStrategy
+    elif snapshot.is_incremental_unmanaged:
+        klass = IncrementalUnmanagedStrategy
     elif snapshot.is_view:
         klass = ViewStrategy
     else:
@@ -786,6 +789,24 @@ def append(
         )
 
 
+class IncrementalUnmanagedStrategy(MaterializableStrategy):
+    def insert(
+        self,
+        model: Model,
+        name: str,
+        query_or_df: QueryOrDF,
+        snapshots: t.Dict[str, Snapshot],
+        is_dev: bool,
+        **kwargs: t.Any,
+    ) -> None:
+        if isinstance(model.kind, IncrementalUnmanagedKind) and model.kind.insert_overwrite:
+            self.adapter.insert_overwrite(
+                name, query_or_df, columns_to_types=model.columns_to_types
+            )
+        else:
+            self.append(model, name, query_or_df, snapshots, is_dev, **kwargs)
+
+
 class FullRefreshStrategy(MaterializableStrategy):
     def insert(
         self,

sqlmesh/dbt/builtin.py

@@ -145,10 +145,6 @@ def env_var(name: str, default: t.Optional[str] = None) -> t.Optional[str]:
     return os.environ.get(name, default)
 
 
-def is_incremental() -> bool:
-    return False
-
-
 def log(msg: str, info: bool = False) -> str:
     print(msg)
     return ""
@@ -264,7 +260,6 @@ def _try_literal_eval(value: str) -> t.Any:
     "flags": Flags(),
     "fromjson": from_json,
     "fromyaml": from_yaml,
-    "is_incremental": is_incremental,
     "log": no_log,
     "modules": Modules(),
     "print": no_log,
@@ -320,6 +315,9 @@ def create_builtin_globals(
     if variables is not None:
         builtin_globals["var"] = generate_var(variables)
 
+    is_incremental = jinja_globals.pop("has_intervals", False)
+    builtin_globals["is_incremental"] = lambda: is_incremental
+
     builtin_globals["builtins"] = AttributeDict(
         {k: builtin_globals.get(k) for k in ("ref", "source", "config")}
     )

sqlmesh/dbt/model.py

@@ -3,7 +3,7 @@
 import logging
 import typing as t
 
-from pydantic import Field, validator
+from pydantic import validator
 from sqlglot import exp
 from sqlglot.helper import ensure_list
 
@@ -12,6 +12,7 @@
 from sqlmesh.core.model import (
     IncrementalByTimeRangeKind,
     IncrementalByUniqueKeyKind,
+    IncrementalUnmanagedKind,
     Model,
     ModelKind,
     ModelKindName,
@@ -66,7 +67,7 @@ class ModelConfig(BaseModelConfig):
 
     # sqlmesh fields
     sql: SqlStr = SqlStr("")
-    time_column_: t.Optional[str] = Field(None, alias="time_column")
+    time_column: t.Optional[str] = None
     cron: t.Optional[str] = None
     dialect: t.Optional[str] = None
     batch_size: t.Optional[int] = None
@@ -115,22 +116,10 @@ def _validate_partition_by(cls, v: t.Any) -> t.Union[t.List[str], t.Dict[str, t.
         **BaseModelConfig._FIELD_UPDATE_STRATEGY,
         **{
             "sql": UpdateStrategy.IMMUTABLE,
-            "time_column_": UpdateStrategy.IMMUTABLE,
+            "time_column": UpdateStrategy.IMMUTABLE,
         },
     }
 
-    @property
-    def time_column(self) -> t.Optional[str]:
-        if self.time_column_:
-            return self.time_column_
-        if (
-            isinstance(self.partition_by, dict)
-            and self.partition_by["data_type"] != "int64"
-            and self.incremental_strategy in INCREMENTAL_BY_TIME_STRATEGIES
-        ):
-            return self.partition_by["field"]
-        return None
-
     @property
     def model_dialect(self) -> t.Optional[str]:
         return self.dialect or self.meta.get("dialect", None)
@@ -165,7 +154,7 @@ def model_kind(self, target: TargetConfig) -> ModelKind:
                 is_supported = True
                 if strategy not in INCREMENTAL_BY_TIME_STRATEGIES:
                     logger.warning(
-                        "SQLMesh IncrementalByTime is not compatible with '%s' incremental strategy in model '%s'. Supported strategies include %s.",
+                        "SQLMesh incremental by time strategy is not compatible with '%s' incremental strategy in model '%s'. Supported strategies include %s.",
                         strategy,
                         self.sql_name,
                         collection_to_str(INCREMENTAL_BY_TIME_STRATEGIES),
@@ -178,6 +167,7 @@ def model_kind(self, target: TargetConfig) -> ModelKind:
                     forward_only=not is_supported,
                     disable_restatement=not is_supported,
                 )
+
             if self.unique_key:
                 strategy = self.incremental_strategy or target.default_incremental_strategy(
                     IncrementalByUniqueKeyKind
@@ -187,16 +177,21 @@ def model_kind(self, target: TargetConfig) -> ModelKind:
                     and strategy not in INCREMENTAL_BY_UNIQUE_KEY_STRATEGIES
                 ):
                     raise ConfigError(
-                        f"{self.sql_name}: SQLMesh IncrementalByUniqueKey is not compatible with '{strategy}'"
+                        f"{self.sql_name}: SQLMesh incremental by unique key strategy is not compatible with '{strategy}'"
                         f" incremental strategy. Supported strategies include {collection_to_str(INCREMENTAL_BY_UNIQUE_KEY_STRATEGIES)}."
                     )
                 return IncrementalByUniqueKeyKind(unique_key=self.unique_key, **incremental_kwargs)
 
-            raise ConfigError(
-                f"{self.sql_name}: Incremental materialization requires either a "
-                f"time_column ({collection_to_str(INCREMENTAL_BY_TIME_STRATEGIES)}) or a "
-                f"unique_key ({collection_to_str(INCREMENTAL_BY_UNIQUE_KEY_STRATEGIES.union(['none']))}) configuration."
+            logger.warning(
+                "Using unmanaged incremental materialization for model '%s'. Some features might not be available. Consider adding either a time_column (%s) or a unique_key (%s) configuration to mitigate this",
+                self.sql_name,
+                collection_to_str(INCREMENTAL_BY_TIME_STRATEGIES),
+                collection_to_str(INCREMENTAL_BY_UNIQUE_KEY_STRATEGIES.union(["none"])),
+            )
+            strategy = self.incremental_strategy or target.default_incremental_strategy(
+                IncrementalUnmanagedKind
             )
+            return IncrementalUnmanagedKind(insert_overwrite=strategy == "insert_overwrite")
         if materialization == Materialization.EPHEMERAL:
             return ModelKind(name=ModelKindName.EMBEDDED)
         raise ConfigError(f"{materialization.value} materialization not supported.")