Repository: TobikoData/sqlmesh
Pull request #2548 (2024-05-01 17:19:10)

Improve error message when the parsing of 'cluster_by' / 'partition_by' fields fails in the dbt loader
Make the messages more specific about which field caused the failure.


Patch:
diff --git a/docs/concepts/models/python_models.md b/docs/concepts/models/python_models.md
index b9dc9975..780ee80f 100644
--- a/docs/concepts/models/python_models.md
+++ b/docs/concepts/models/python_models.md
@@ -41,29 +41,32 @@ If the function output is too large, it can also be returned in chunks using Pyt
 
 The arguments provided in the `@model` specification have the same names as those provided in a SQL model's `MODEL` DDL.
 
-Most of the arguments are simply Python-formatted equivalents of the SQL version, but Python model `kind`s are specified with model kind objects. All model kind arguments are listed in the [models configuration reference page](../../reference/model_configuration.md#model-kind-properties). A model's `kind` object must be imported at the beginning of the model definition file before use in the model specification.
+Python model `kind`s are specified with a Python dictionary containing the kind's name and arguments. All model kind arguments are listed in the [models configuration reference page](../../reference/model_configuration.md#model-kind-properties).
 
-Supported model kind objects include:
+The model `kind` dictionary must contain a `name` key whose value is a member of the [`ModelKindName` enum class](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#ModelKindName). The `ModelKindName` class must be imported at the beginning of the model definition file before being used in the `@model` specification.
 
-- [ViewKind()](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#ViewKind)
-- [FullKind()](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#FullKind)
-- [SeedKind()](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#SeedKind)
-- [IncrementalByTimeRangeKind()](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#IncrementalByTimeRangeKind)
-- [IncrementalByUniqueKeyKind()](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#IncrementalByUniqueKeyKind)
-- [SCDType2KindByTimeKind()](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#SCDType2ByTimeKind)
-- [SCDType2KindByColumnKind()](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#SCDType2ByColumnKind)
-- [EmbeddedKind()](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#EmbeddedKind)
-- [ExternalKind()](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#ExternalKind)
+Supported `kind` dictionary `name` values are:
+
+- `ModelKindName.VIEW`
+- `ModelKindName.FULL`
+- `ModelKindName.SEED`
+- `ModelKindName.INCREMENTAL_BY_TIME_RANGE`
+- `ModelKindName.INCREMENTAL_BY_UNIQUE_KEY`
+- `ModelKindName.SCD_TYPE_2_BY_TIME`
+- `ModelKindName.SCD_TYPE_2_BY_COLUMN`
+- `ModelKindName.EMBEDDED`
+- `ModelKindName.EXTERNAL`
 
 This example demonstrates how to specify an incremental by time range model kind in Python:
 
 ```python linenums="1"
 from sqlmesh import ExecutionContext, model
-from sqlmesh.core.model import IncrementalByTimeRangeKind
+from sqlmesh.core.model.kind import ModelKindName
 
 @model(
     "docs_example.incremental_model",
-    kind=IncrementalByTimeRangeKind(
+    kind=dict(
+        name=ModelKindName.INCREMENTAL_BY_TIME_RANGE,
         time_column="model_time_column"
     )
 )
diff --git a/docs/guides/configuration.md b/docs/guides/configuration.md
index be1af2a7..b3945623 100644
--- a/docs/guides/configuration.md
+++ b/docs/guides/configuration.md
@@ -495,7 +495,7 @@ These pages describe the connection configuration options for each execution eng
 
 Configuration for the state backend connection if different from the data warehouse connection.
 **Using the same connection for data warehouse and state is only recommended for non-production deployments of SQLMesh.**
-Unlike data transformations, storing state information requires database transactions. Data warehouses aren’t optimized for executing transactions, so storing state information in them can slow down your project. 
+Unlike data transformations, storing state information requires database transactions. Data warehouses aren’t optimized for executing transactions, so storing state information in them can slow down your project.
 Even worse data corruption can occur with simultaneous writes to the same table.
 Therefore, using your data warehouse is fine for testing but once you start running SQLMesh in production, you should use a dedicated state connection.
 
@@ -937,11 +937,12 @@ This example demonstrates how to specify an incremental by time range model kind
 
     ```python linenums="1"
     from sqlmesh import ExecutionContext, model
-    from sqlmesh.core.model import IncrementalByTimeRangeKind
+    from sqlmesh.core.model.kind import ModelKindName
 
     @model(
         "docs_example.incremental_model",
-        kind=IncrementalByTimeRangeKind(
+        kind=dict(
+            name=ModelKindName.INCREMENTAL_BY_TIME_RANGE,
             time_column="ds"
         )
     )
diff --git a/docs/reference/model_configuration.md b/docs/reference/model_configuration.md
index df5e3cd8..7678ef11 100644
--- a/docs/reference/model_configuration.md
+++ b/docs/reference/model_configuration.md
@@ -60,13 +60,13 @@ Configuration options for models of the [`VIEW` kind](../concepts/models/model_k
 | -------------- | ---------------------------------------------------------------------------------------------------- | :--: | :------: |
 | `materialized` | Whether views should be materialized (for engines supporting materialized views). (Default: `False`) | bool |    N     |
 
-Python model configuration object: [ViewKind()](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#ViewKind)
+Python model kind `name` enum value: [ModelKindName.VIEW](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#ModelKindName)
 
 ### `FULL` models
 
 The [`FULL` model kind](../concepts/models/model_kinds.md#full) does not support any configuration options other than the [general model properties listed above](#general-model-properties).
 
-Python model configuration object: [FullKind()](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#FullKind)
+Python model kind `name` enum value: [ModelKindName.FULL](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#ModelKindName)
 
 ### Incremental models
 
@@ -89,7 +89,7 @@ Configuration options for [`INCREMENTAL_BY_TIME_RANGE` models](../concepts/model
 | `forward_only`        | Whether the model's changes should always be classified as [forward-only](../concepts/plans.md#forward-only-change). (Default: `False`) | bool |    N     |
 | `disable_restatement` | Whether [restatements](../concepts/plans.md#restatement-plans) should be disabled for the model. (Default: `False`)                     | bool |    N     |
 
-Python model configuration object: [IncrementalByTimeRangeKind()](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#IncrementalByTimeRangeKind)
+Python model kind `name` enum value: [ModelKindName.INCREMENTAL_BY_TIME_RANGE](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#ModelKindName)
 
 #### Incremental by unique key
 
@@ -102,7 +102,7 @@ Configuration options for [`INCREMENTAL_BY_UNIQUE_KEY` models](../concepts/model
 | `forward_only`        | Whether the model's changes should always be classified as [forward-only](../concepts/plans.md#forward-only-change). (Default: `False`) |       bool       |    N     |
 | `disable_restatement` | Whether [restatements](../concepts/plans.md#restatement-plans) should be disabled for the model. (Default: `False`)                     |       bool       |    N     |
 
-Python model configuration object: [IncrementalByUniqueKeyKind()](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#IncrementalByUniqueKeyKind)
+Python model kind `name` enum value: [ModelKindName.INCREMENTAL_BY_UNIQUE_KEY](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#ModelKindName)
 
 #### SCD Type 2 models
 
@@ -124,6 +124,8 @@ Configuration options for [`SCD_TYPE_2_BY_TIME` models](../concepts/models/model
 | `updated_at_name`          | The model column containing each row's updated at date. (Default: `updated_at`)                                                                                                  | str  |    N     |
 | `updated_at_as_valid_from` | By default, for new rows the `valid_from` column is set to 1970-01-01 00:00:00. This sets `valid_from` to the value of `updated_at` when the row is inserted. (Default: `False`) | bool |    N     |
 
+Python model kind `name` enum value: [ModelKindName.SCD_TYPE_2_BY_TIME](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#ModelKindName)
+
 ##### SCD Type 2 By Column
 
 Configuration options for [`SCD_TYPE_2_BY_COLUMN` models](../concepts/models/model_kinds.md#scd-type-2) (in addition to [general model properties](#general-model-properties), [incremental model properties](#incremental-models), and [SCD Type 2 properties](#scd-type-2-models)).
@@ -133,7 +135,7 @@ Configuration options for [`SCD_TYPE_2_BY_COLUMN` models](../concepts/models/mod
 | `columns`                      | Columns whose changed data values indicate a data update (instead of an `updated_at` column). `*` to represent that all columns should be checked.                          | str \| array[str] |    Y     |
 | `execution_time_as_valid_from` | By default, for new rows `valid_from` is set to 1970-01-01 00:00:00. This changes the behavior to set it to the execution_time of when the pipeline ran. (Default: `False`) |       bool        |    N     |
 
-Python model configuration object: [SCDType2Kind()](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#SCDType2Kind)
+Python model kind `name` enum value: [ModelKindName.SCD_TYPE_2_BY_COLUMN](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#ModelKindName)
 
 ### `SEED` models
 
@@ -172,4 +174,4 @@ Options specified within the `kind` property's `csv_settings` property (override
 | `lineterminator`   | Character used to denote a line break. More information at the [Pandas documentation](https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.read_csv.html).                                                                                                             | str  | N        |
 | `encoding`         | Encoding to use for UTF when reading/writing (ex. 'utf-8'). More information at the [Pandas documentation](https://pandas.pydata.org/pandas-docs/stable/reference/api/pandas.read_csv.html).                                                                                        | str  | N        |
 
-Python model configuration object: [SeedKind()](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#SeedKind)
+Python model kind `name` enum value: [ModelKindName.SEED](https://sqlmesh.readthedocs.io/en/stable/_readthedocs/html/sqlmesh/core/model/kind.html#ModelKindName)
diff --git a/examples/sushi/models/items.py b/examples/sushi/models/items.py
index 4e855d70..d20ea45d 100644
--- a/examples/sushi/models/items.py
+++ b/examples/sushi/models/items.py
@@ -8,7 +8,7 @@ from helper import iter_dates  # type: ignore
 from sqlglot.expressions import to_column
 
 from sqlmesh import ExecutionContext, model
-from sqlmesh.core.model import IncrementalByTimeRangeKind
+from sqlmesh.core.model.kind import ModelKindName
 from sqlmesh.utils.date import to_date
 
 ITEMS = [
@@ -49,7 +49,9 @@ ITEMS = [
 
 @model(
     "sushi.items",
-    kind=IncrementalByTimeRangeKind(time_column="event_date", batch_size=30),
+    kind=dict(
+        name=ModelKindName.INCREMENTAL_BY_TIME_RANGE, time_column="event_date", batch_size=30
+    ),
     start="1 week ago",
     cron="@daily",
     columns={
diff --git a/examples/sushi/models/order_items.py b/examples/sushi/models/order_items.py
index b247809d..b04b280c 100644
--- a/examples/sushi/models/order_items.py
+++ b/examples/sushi/models/order_items.py
@@ -9,7 +9,7 @@ from sqlglot import exp
 from sqlglot.expressions import to_column
 
 from sqlmesh import ExecutionContext, model
-from sqlmesh.core.model import IncrementalByTimeRangeKind
+from sqlmesh.core.model.kind import ModelKindName
 from sqlmesh.utils.date import to_date, to_ds
 
 ITEMS = "sushi.items"
@@ -21,7 +21,8 @@ def get_items_table(context: ExecutionContext) -> str:
 
 @model(
     "sushi.order_items",
-    kind=IncrementalByTimeRangeKind(
+    kind=dict(
+        name=ModelKindName.INCREMENTAL_BY_TIME_RANGE,
         time_column="event_date",
         batch_size=30,
     ),
diff --git a/examples/sushi/models/orders.py b/examples/sushi/models/orders.py
index d8e06225..75b0b25b 100644
--- a/examples/sushi/models/orders.py
+++ b/examples/sushi/models/orders.py
@@ -6,7 +6,7 @@ import pandas as pd
 from helper import iter_dates  # type: ignore
 
 from sqlmesh import ExecutionContext, model
-from sqlmesh.core.model import IncrementalByTimeRangeKind
+from sqlmesh.core.model.kind import ModelKindName
 from sqlmesh.utils.date import to_date
 
 CUSTOMERS = list(range(0, 100))
@@ -16,7 +16,9 @@ WAITERS = list(range(0, 10))
 @model(
     "sushi.orders",
     description="Table of sushi orders.",
-    kind=IncrementalByTimeRangeKind(time_column="event_date", batch_size=30),
+    kind=dict(
+        name=ModelKindName.INCREMENTAL_BY_TIME_RANGE, time_column="event_date", batch_size=30
+    ),
     start="1 week ago",
     cron="@daily",
     grains=[
diff --git a/examples/sushi/models/raw_marketing.py b/examples/sushi/models/raw_marketing.py
index 6a8df6d1..3f47866d 100644
--- a/examples/sushi/models/raw_marketing.py
+++ b/examples/sushi/models/raw_marketing.py
@@ -8,13 +8,13 @@ from helper import iter_dates  # type: ignore
 from sqlglot import exp
 
 from sqlmesh import ExecutionContext, model
-from sqlmesh.core.model import FullKind
+from sqlmesh.core.model.kind import ModelKindName
 
 
 @model(
     "sushi.raw_marketing",
     description="Table of marketing status.",
-    kind=FullKind(),
+    kind=dict(name=ModelKindName.FULL),
     start="1 week ago",
     cron="@daily",
     grains=[
diff --git a/examples/sushi/models/waiters.py b/examples/sushi/models/waiters.py
index d8668a03..56d13880 100644
--- a/examples/sushi/models/waiters.py
+++ b/examples/sushi/models/waiters.py
@@ -3,13 +3,13 @@ from sqlglot import exp
 
 from sqlmesh import model
 from sqlmesh.core.macros import MacroEvaluator
-from sqlmesh.core.model import EmbeddedKind
+from sqlmesh.core.model.kind import ModelKindName
 
 
 @model(
     "sushi.waiters",
     is_sql=True,
-    kind=EmbeddedKind(),
+    kind=dict(name=ModelKindName.EMBEDDED),
     owner="jen",
     cron="@daily",
 )
diff --git a/sqlmesh/core/config/connection.py b/sqlmesh/core/config/connection.py
index 66756e8f..aeeaf275 100644
--- a/sqlmesh/core/config/connection.py
+++ b/sqlmesh/core/config/connection.py
@@ -8,6 +8,7 @@ import pathlib
 import sys
 import typing as t
 from enum import Enum
+from functools import partial
 
 from pydantic import Field
 from sqlglot import exp
@@ -84,15 +85,21 @@ class ConnectionConfig(abc.ABC, BaseConfig):
         """Whether this connection is recommended for being used as a state sync for production state syncs"""
         return self.type_ in RECOMMENDED_STATE_SYNC_ENGINES
 
+    @property
+    def _connection_factory_with_kwargs(self) -> t.Callable[[], t.Any]:
+        """A function that is called to return a connection object for the given Engine Adapter"""
+        return partial(
+            self._connection_factory,
+            **{
+                **self._static_connection_kwargs,
+                **{k: v for k, v in self.dict().items() if k in self._connection_kwargs_keys},
+            },
+        )
+
     def create_engine_adapter(self, register_comments_override: bool = False) -> EngineAdapter:
         """Returns a new instance of the Engine Adapter."""
         return self._engine_adapter(
-            lambda: self._connection_factory(
-                **{
-                    **self._static_connection_kwargs,
-                    **{k: v for k, v in self.dict().items() if k in self._connection_kwargs_keys},
-                }
-            ),
+            self._connection_factory_with_kwargs,
             multithreaded=self.concurrent_tasks > 1,
             cursor_kwargs=self._cursor_kwargs,
             default_catalog=self.get_catalog(),
diff --git a/sqlmesh/core/model/decorator.py b/sqlmesh/core/model/decorator.py
index 3320fb35..55b5f6c7 100644
--- a/sqlmesh/core/model/decorator.py
+++ b/sqlmesh/core/model/decorator.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 import typing as t
 from pathlib import Path
 
@@ -9,10 +10,13 @@ from sqlglot.dialects.dialect import DialectType
 from sqlmesh.core import constants as c
 from sqlmesh.core.dialect import MacroFunc
 from sqlmesh.core.model.definition import Model, create_python_model, create_sql_model
+from sqlmesh.core.model.kind import ModelKindName, _ModelKind
 from sqlmesh.utils import registry_decorator
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.metaprogramming import build_env, serialize_env
 
+logger = logging.getLogger(__name__)
+
 
 class model(registry_decorator):
     """Specifies a function is a python based model."""
@@ -27,6 +31,18 @@ class model(registry_decorator):
         if not is_sql and "columns" not in kwargs:
             raise ConfigError("Python model must define column schema.")
 
+        kind = kwargs.get("kind", None)
+        if kind is not None:
+            if isinstance(kind, _ModelKind):
+                logger.warning(
+                    f"""Python model "{name}"'s `kind` argument was passed a SQLMesh `{type(kind).__name__}` object. This may result in unexpected behavior - provide a dictionary instead."""
+                )
+            elif isinstance(kind, dict):
+                if not "name" in kind or not isinstance(kind.get("name"), ModelKindName):
+                    raise ConfigError(
+                        f"""Python model "{name}"'s `kind` dictionary must contain a `name` key with a valid ModelKindName enum value."""
+                    )
+
         self.name = name
         self.is_sql = is_sql
         self.kwargs = kwargs
diff --git a/sqlmesh/core/state_sync/base.py b/sqlmesh/core/state_sync/base.py
index 6c9fe570..f603da36 100644
--- a/sqlmesh/core/state_sync/base.py
+++ b/sqlmesh/core/state_sync/base.py
@@ -18,7 +18,7 @@ from sqlmesh.core.snapshot import (
     SnapshotTableCleanupTask,
     SnapshotTableInfo,
 )
-from sqlmesh.core.snapshot.definition import Interval
+from sqlmesh.core.snapshot.definition import Interval, SnapshotIntervals
 from sqlmesh.utils import major_minor
 from sqlmesh.utils.date import TimeLike
 from sqlmesh.utils.errors import SQLMeshError
@@ -303,26 +303,6 @@ class StateSync(StateReader, abc.ABC):
             name: The name of the environment to invalidate.
         """
 
-    @abc.abstractmethod
-    def add_interval(
-        self,
-        snapshot: Snapshot,
-        start: TimeLike,
-        end: TimeLike,
-        is_dev: bool = False,
-    ) -> None:
-        """Add an interval to a snapshot and sync it to the store.
-
-        Snapshots must be pushed before adding intervals to them.
-
-        Args:
-            snapshot: The snapshot like object to add an interval to.
-            start: The start of the interval to add.
-            end: The end of the interval to add.
-            is_dev: Indicates whether the given interval is being added while in
-                development mode.is_dev.
-        """
-
     @abc.abstractmethod
     def remove_interval(
         self,
@@ -429,6 +409,42 @@ class StateSync(StateReader, abc.ABC):
     def rollback(self) -> None:
         """Rollback to previous backed up state."""
 
+    @abc.abstractmethod
+    def _add_snapshot_intervals(self, snapshot_intervals: SnapshotIntervals) -> None:
+        """Add snapshot intervals to state
+
+        Args:
+            snapshot_intervals: The snapshot intervals to add.
+        """
+
+    def add_interval(
+        self,
+        snapshot: Snapshot,
+        start: TimeLike,
+        end: TimeLike,
+        is_dev: bool = False,
+    ) -> None:
+        """Add an interval to a snapshot and sync it to the store.
+
+        Args:
+            snapshot: The snapshot like object to add an interval to.
+            start: The start of the interval to add.
+            end: The end of the interval to add.
+            is_dev: Indicates whether the given interval is being added while in development mode
+        """
+        start_ts, end_ts = snapshot.inclusive_exclusive(start, end, strict=False)
+        if not snapshot.version:
+            raise SQLMeshError("Snapshot version must be set to add an interval.")
+        intervals = [(start_ts, end_ts)]
+        snapshot_intervals = SnapshotIntervals(
+            name=snapshot.name,
+            identifier=snapshot.identifier,
+            version=snapshot.version,
+            intervals=intervals if not is_dev else [],
+            dev_intervals=intervals if is_dev else [],
+        )
+        self._add_snapshot_intervals(snapshot_intervals)
+
 
 class DelegatingStateSync(StateSync):
     def __init__(self, state_sync: StateSync) -> None:
diff --git a/sqlmesh/core/state_sync/cache.py b/sqlmesh/core/state_sync/cache.py
index ef92dc98..209b67e5 100644
--- a/sqlmesh/core/state_sync/cache.py
+++ b/sqlmesh/core/state_sync/cache.py
@@ -11,7 +11,7 @@ from sqlmesh.core.snapshot import (
     SnapshotInfoLike,
     SnapshotTableCleanupTask,
 )
-from sqlmesh.core.snapshot.definition import Interval
+from sqlmesh.core.snapshot.definition import Interval, SnapshotIntervals
 from sqlmesh.core.state_sync.base import DelegatingStateSync, StateSync
 from sqlmesh.utils.date import TimeLike, now_timestamp
 
@@ -129,15 +129,9 @@ class CachingStateSync(DelegatingStateSync):
         self.snapshot_cache.clear()
         return self.state_sync.delete_expired_snapshots()
 
-    def add_interval(
-        self,
-        snapshot: Snapshot,
-        start: TimeLike,
-        end: TimeLike,
-        is_dev: bool = False,
-    ) -> None:
-        self.snapshot_cache.pop(snapshot.snapshot_id, None)
-        self.state_sync.add_interval(snapshot, start, end, is_dev)
+    def _add_snapshot_intervals(self, snapshot_intervals: SnapshotIntervals) -> None:
+        self.snapshot_cache.pop(snapshot_intervals.snapshot_id, None)
+        self.state_sync._add_snapshot_intervals(snapshot_intervals)
 
     def remove_interval(
         self,
diff --git a/sqlmesh/core/state_sync/engine_adapter.py b/sqlmesh/core/state_sync/engine_adapter.py
index 3f50487f..b1bea4ac 100644
--- a/sqlmesh/core/state_sync/engine_adapter.py
+++ b/sqlmesh/core/state_sync/engine_adapter.py
@@ -624,27 +624,48 @@ class EngineAdapterStateSync(CommonStateSyncMixin, StateSync):
         end: TimeLike,
         is_dev: bool = False,
     ) -> None:
-        start_ts, end_ts = snapshot.inclusive_exclusive(start, end, strict=False)
-        if start_ts >= end_ts:
-            logger.info(
-                "Skipping partial interval (%s, %s) for snapshot %s",
-                start,
-                end,
-                snapshot.snapshot_id,
-            )
-            return
+        super().add_interval(snapshot, start, end, is_dev)
 
-        logger.info(
-            "Adding %s (%s, %s) for snapshot %s",
-            "dev interval" if is_dev else "interval",
-            start_ts,
-            end_ts,
-            snapshot.snapshot_id,
-        )
+    @transactional()
+    def _add_snapshot_intervals(self, snapshot_intervals: SnapshotIntervals) -> None:
+        def remove_partial_intervals(
+            intervals: t.List[Interval], snapshot_id: SnapshotId, *, is_dev: bool
+        ) -> t.List[Interval]:
+            results = []
+            for start_ts, end_ts in intervals:
+                if start_ts < end_ts:
+                    logger.info(
+                        "Adding %s (%s, %s) for snapshot %s",
+                        "dev interval" if is_dev else "interval",
+                        start_ts,
+                        end_ts,
+                        snapshot_id,
+                    )
+                    results.append((start_ts, end_ts))
+                else:
+                    logger.info(
+                        "Skipping partial interval (%s, %s) for snapshot %s",
+                        start_ts,
+                        end_ts,
+                        snapshot_intervals.snapshot_id,
+                    )
+            return results
 
+        snapshot_intervals = snapshot_intervals.copy(
+            update={
+                "intervals": remove_partial_intervals(
+                    snapshot_intervals.intervals, snapshot_intervals.snapshot_id, is_dev=False
+                ),
+                "dev_intervals": remove_partial_intervals(
+                    snapshot_intervals.dev_intervals, snapshot_intervals.snapshot_id, is_dev=True
+                ),
+            }
+        )
+        if not snapshot_intervals.intervals and not snapshot_intervals.dev_intervals:
+            return
         self.engine_adapter.insert_append(
             self.intervals_table,
-            _intervals_to_df([(snapshot, (start_ts, end_ts))], is_dev, False),
+            _snapshot_interval_to_df(snapshot_intervals, is_removed=False),
             columns_to_types=self._interval_columns_to_types,
         )
 
@@ -1312,6 +1333,27 @@ def _intervals_to_df(
     )
 
 
+def _snapshot_interval_to_df(
+    snapshot_intervals: SnapshotIntervals,
+    is_removed: bool = False,
+) -> pd.DataFrame:
+    return pd.DataFrame(
+        [
+            _interval_to_df(
+                snapshot_intervals,
+                start_ts,
+                end_ts,
+                is_dev=is_dev,
+                is_removed=is_removed,
+            )
+            for is_dev in (False, True)
+            for start_ts, end_ts in getattr(
+                snapshot_intervals, "dev_intervals" if is_dev else "intervals"
+            )
+        ]
+    )
+
+
 def _interval_to_df(
     snapshot: t.Union[SnapshotInfoLike, SnapshotIntervals],
     start_ts: int,
diff --git a/sqlmesh/dbt/model.py b/sqlmesh/dbt/model.py
index e4cba5c7..721c7dd4 100644
--- a/sqlmesh/dbt/model.py
+++ b/sqlmesh/dbt/model.py
@@ -4,6 +4,7 @@ import logging
 import typing as t
 
 from sqlglot import exp
+from sqlglot.errors import SqlglotError
 from sqlglot.helper import ensure_list
 
 from sqlmesh.core import dialect as d
@@ -300,7 +301,12 @@ class ModelConfig(BaseModelConfig):
     def _big_query_partition_by_expr(self, context: DbtContext) -> exp.Expression:
         assert isinstance(self.partition_by, dict)
         data_type = self.partition_by["data_type"].lower()
-        field = d.parse_one(self.partition_by["field"], dialect="bigquery")
+        raw_field = self.partition_by["field"]
+        try:
+            field = d.parse_one(raw_field, dialect="bigquery")
+        except SqlglotError as e:
+            raise ConfigError(f"Failed to parse partition_by field '{raw_field}': {e}") from e
+
         if data_type == "date" and self.partition_by["granularity"].lower() == "day":
             return field
 
@@ -346,9 +352,13 @@ class ModelConfig(BaseModelConfig):
             )
 
         if self.cluster_by:
-            optional_kwargs["clustered_by"] = [
-                d.parse_one(c, dialect=dialect).name for c in self.cluster_by
-            ]
+            clustered_by = []
+            for c in self.cluster_by:
+                try:
+                    clustered_by.append(d.parse_one(c, dialect=dialect).name)
+                except SqlglotError as e:
+                    raise ConfigError(f"Failed to parse cluster_by field '{c}': {e}") from e
+            optional_kwargs["clustered_by"] = clustered_by
 
         model_kwargs = self.sqlmesh_model_kwargs(context)
         if self.sql_header:
diff --git a/sqlmesh/schedulers/airflow/state_sync.py b/sqlmesh/schedulers/airflow/state_sync.py
index 6a65348b..33d58ea1 100644
--- a/sqlmesh/schedulers/airflow/state_sync.py
+++ b/sqlmesh/schedulers/airflow/state_sync.py
@@ -12,7 +12,7 @@ from sqlmesh.core.snapshot import (
     SnapshotInfoLike,
     SnapshotTableCleanupTask,
 )
-from sqlmesh.core.snapshot.definition import Interval
+from sqlmesh.core.snapshot.definition import Interval, SnapshotIntervals
 from sqlmesh.core.state_sync import StateSync, Versions
 from sqlmesh.core.state_sync.base import PromotionResult
 from sqlmesh.schedulers.airflow.client import AirflowClient
@@ -219,6 +219,9 @@ class HttpStateSync(StateSync):
         """
         raise NotImplementedError("Adding intervals is not supported by the Airflow state sync.")
 
+    def _add_snapshot_intervals(self, snapshot_intervals: SnapshotIntervals) -> None:
+        raise NotImplementedError("Adding intervals is not supported by the Airflow state sync.")
+
     def remove_interval(
         self,
         snapshot_intervals: t.Sequence[t.Tuple[SnapshotInfoLike, Interval]],


Test patch:
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index 7ae481f4..6d6b9d23 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -19,6 +19,7 @@ from sqlmesh.core.context import Context, ExecutionContext
 from sqlmesh.core.dialect import parse
 from sqlmesh.core.macros import MacroEvaluator, macro
 from sqlmesh.core.model import (
+    FullKind,
     IncrementalByTimeRangeKind,
     IncrementalUnmanagedKind,
     ModelCache,
@@ -26,6 +27,7 @@ from sqlmesh.core.model import (
     SeedKind,
     SqlModel,
     TimeColumn,
+    ViewKind,
     create_external_model,
     create_seed_model,
     create_sql_model,
@@ -33,7 +35,7 @@ from sqlmesh.core.model import (
     model,
 )
 from sqlmesh.core.model.common import parse_expression
-from sqlmesh.core.model.kind import _model_kind_validator
+from sqlmesh.core.model.kind import ModelKindName, _model_kind_validator
 from sqlmesh.core.model.seed import CsvSettings
 from sqlmesh.core.node import IntervalUnit, _Node
 from sqlmesh.core.snapshot import Snapshot, SnapshotChangeCategory
@@ -1616,6 +1618,76 @@ def test_python_models_returning_sql(assert_exp_eq) -> None:
     )
 
 
+def test_python_model_decorator_kind() -> None:
+    logger = logging.getLogger("sqlmesh.core.model.decorator")
+
+    # no kind specified -> default View kind
+    @model("default_kind", columns={'"COL"': "int"})
+    def a_model(context):
+        pass
+
+    python_model = model.get_registry()["default_kind"].model(
+        module_path=Path("."),
+        path=Path("."),
+    )
+
+    assert isinstance(python_model.kind, ViewKind)
+
+    # string kind name specified
+    @model("kind_string", kind="full", columns={'"COL"': "int"})
+    def b_model(context):
+        pass
+
+    python_model = model.get_registry()["kind_string"].model(
+        module_path=Path("."),
+        path=Path("."),
+    )
+
+    assert isinstance(python_model.kind, FullKind)
+
+    # error if kind dict with no `name` key
+    with pytest.raises(ConfigError, match="`kind` dictionary must contain a `name` key"):
+
+        @model("kind_empty_dict", kind=dict(), columns={'"COL"': "int"})
+        def my_model(context):
+            pass
+
+    # error if kind dict with `name` key whose type is not a ModelKindName enum
+    with pytest.raises(ConfigError, match="with a valid ModelKindName enum value"):
+
+        @model("kind_dict_badname", kind=dict(name="test"), columns={'"COL"': "int"})
+        def my_model(context):
+            pass
+
+    # warning if kind is ModelKind instance
+    with patch.object(logger, "warning") as mock_logger:
+
+        @model("kind_instance", kind=FullKind(), columns={'"COL"': "int"})
+        def my_model(context):
+            pass
+
+        assert (
+            mock_logger.call_args[0][0]
+            == """Python model "kind_instance"'s `kind` argument was passed a SQLMesh `FullKind` object. This may result in unexpected behavior - provide a dictionary instead."""
+        )
+
+    # no warning with valid kind dict
+    with patch.object(logger, "warning") as mock_logger:
+
+        @model("kind_valid_dict", kind=dict(name=ModelKindName.FULL), columns={'"COL"': "int"})
+        def my_model(context):
+            pass
+
+        python_model = model.get_registry()["kind_valid_dict"].model(
+            module_path=Path("."),
+            path=Path("."),
+        )
+
+        assert isinstance(python_model.kind, FullKind)
+
+        assert not mock_logger.call_args
+
+
 def test_star_expansion(assert_exp_eq) -> None:
     context = Context(config=Config())