Repository: TobikoData/sqlmesh
Pull request #3120 (2024-09-11 19:29:08)

Chore: Refactor the scheduler API to support running of precomputed missing intervals


Patch:
diff --git a/sqlmesh/core/engine_adapter/base.py b/sqlmesh/core/engine_adapter/base.py
index 7e6f0489..f16b8331 100644
--- a/sqlmesh/core/engine_adapter/base.py
+++ b/sqlmesh/core/engine_adapter/base.py
@@ -895,6 +895,7 @@ class EngineAdapter:
         columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
         replace: bool = True,
         materialized: bool = False,
+        materialized_properties: t.Optional[t.Dict[str, t.Any]] = None,
         table_description: t.Optional[str] = None,
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         view_properties: t.Optional[t.Dict[str, exp.Expression]] = None,
@@ -911,11 +912,15 @@ class EngineAdapter:
             columns_to_types: Columns to use in the view statement.
             replace: Whether or not to replace an existing view defaults to True.
             materialized: Whether to create a a materialized view. Only used for engines that support this feature.
+            materialized_properties: Optional materialized view properties to add to the view.
             table_description: Optional table description from MODEL DDL.
             column_descriptions: Optional column descriptions from model query.
             view_properties: Optional view properties to add to the view.
             create_kwargs: Additional kwargs to pass into the Create expression
         """
+        if materialized_properties and not materialized:
+            raise SQLMeshError("Materialized properties are only supported for materialized views")
+
         if isinstance(query_or_df, pd.DataFrame):
             values: t.List[t.Tuple[t.Any, ...]] = list(
                 query_or_df.itertuples(index=False, name=None)
@@ -955,6 +960,23 @@ class EngineAdapter:
         if not self.SUPPORTS_VIEW_SCHEMA and isinstance(schema, exp.Schema):
             schema = schema.this
 
+        if materialized_properties:
+            partitioned_by = materialized_properties.pop("partitioned_by", None)
+            clustered_by = materialized_properties.pop("clustered_by", None)
+            if partitioned_by and (
+                partitioned_by_prop := self._build_partitioned_by_exp(
+                    partitioned_by, **materialized_properties
+                )
+            ):
+                materialized_properties["catalog_name"] = exp.to_table(view_name).catalog
+                properties.append("expressions", partitioned_by_prop)
+            if clustered_by and (
+                clustered_by_prop := self._build_clustered_by_exp(
+                    clustered_by, **materialized_properties
+                )
+            ):
+                properties.append("expressions", clustered_by_prop)
+
         create_view_properties = self._build_view_properties_exp(
             view_properties,
             (
@@ -2015,6 +2037,24 @@ class EngineAdapter:
             for key, value in table_or_view_properties.items()
         ]
 
+    def _build_partitioned_by_exp(
+        self,
+        partitioned_by: t.List[exp.Expression],
+        *,
+        partition_interval_unit: t.Optional[IntervalUnit] = None,
+        columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
+        catalog_name: t.Optional[str] = None,
+        **kwargs: t.Any,
+    ) -> t.Optional[t.Union[exp.PartitionedByProperty, exp.Property]]:
+        return None
+
+    def _build_clustered_by_exp(
+        self,
+        clustered_by: t.List[str],
+        **kwargs: t.Any,
+    ) -> t.Optional[exp.Cluster]:
+        return None
+
     def _build_table_properties_exp(
         self,
         catalog_name: t.Optional[str] = None,
diff --git a/sqlmesh/core/engine_adapter/base_postgres.py b/sqlmesh/core/engine_adapter/base_postgres.py
index 7b49982d..97b2523c 100644
--- a/sqlmesh/core/engine_adapter/base_postgres.py
+++ b/sqlmesh/core/engine_adapter/base_postgres.py
@@ -92,6 +92,7 @@ class BasePostgresEngineAdapter(EngineAdapter):
         columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
         replace: bool = True,
         materialized: bool = False,
+        materialized_properties: t.Optional[t.Dict[str, t.Any]] = None,
         table_description: t.Optional[str] = None,
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         view_properties: t.Optional[t.Dict[str, exp.Expression]] = None,
@@ -113,6 +114,7 @@ class BasePostgresEngineAdapter(EngineAdapter):
                 columns_to_types=columns_to_types,
                 replace=False,
                 materialized=materialized,
+                materialized_properties=materialized_properties,
                 table_description=table_description,
                 column_descriptions=column_descriptions,
                 view_properties=view_properties,
diff --git a/sqlmesh/core/engine_adapter/bigquery.py b/sqlmesh/core/engine_adapter/bigquery.py
index 93a9a761..2324bbdb 100644
--- a/sqlmesh/core/engine_adapter/bigquery.py
+++ b/sqlmesh/core/engine_adapter/bigquery.py
@@ -9,7 +9,7 @@ from sqlglot import exp
 from sqlglot.transforms import remove_precision_parameterized_types
 
 from sqlmesh.core.dialect import to_schema
-from sqlmesh.core.engine_adapter.mixins import InsertOverwriteWithMergeMixin
+from sqlmesh.core.engine_adapter.mixins import InsertOverwriteWithMergeMixin, ClusteredByMixin
 from sqlmesh.core.engine_adapter.shared import (
     CatalogSupport,
     DataObject,
@@ -38,7 +38,7 @@ logger = logging.getLogger(__name__)
 
 
 @set_catalog()
-class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
+class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin, ClusteredByMixin):
     """
     BigQuery Engine Adapter using the `google-cloud-bigquery` library's DB API.
     """
@@ -553,6 +553,49 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
             value=exp.Literal.string(trunc_method(description)),
         )
 
+    def _build_partitioned_by_exp(
+        self,
+        partitioned_by: t.List[exp.Expression],
+        *,
+        partition_interval_unit: t.Optional[IntervalUnit] = None,
+        columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
+        **kwargs: t.Any,
+    ) -> t.Optional[exp.PartitionedByProperty]:
+        if len(partitioned_by) > 1:
+            raise SQLMeshError("BigQuery only supports partitioning by a single column")
+
+        this = partitioned_by[0]
+        if (
+            isinstance(this, exp.Column)
+            and partition_interval_unit is not None
+            and not partition_interval_unit.is_minute
+        ):
+            column_type: t.Optional[exp.DataType] = (columns_to_types or {}).get(this.name)
+
+            if column_type == exp.DataType.build(
+                "date", dialect=self.dialect
+            ) and partition_interval_unit in (
+                IntervalUnit.MONTH,
+                IntervalUnit.YEAR,
+            ):
+                trunc_func = "DATE_TRUNC"
+            elif column_type == exp.DataType.build("timestamp", dialect=self.dialect):
+                trunc_func = "TIMESTAMP_TRUNC"
+            elif column_type == exp.DataType.build("datetime", dialect=self.dialect):
+                trunc_func = "DATETIME_TRUNC"
+            else:
+                trunc_func = ""
+
+            if trunc_func:
+                this = exp.func(
+                    trunc_func,
+                    this,
+                    exp.var(partition_interval_unit.value.upper()),
+                    dialect=self.dialect,
+                )
+
+        return exp.PartitionedByProperty(this=this)
+
     def _build_table_properties_exp(
         self,
         catalog_name: t.Optional[str] = None,
@@ -567,45 +610,17 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin):
     ) -> t.Optional[exp.Properties]:
         properties: t.List[exp.Expression] = []
 
-        if partitioned_by:
-            if len(partitioned_by) > 1:
-                raise SQLMeshError("BigQuery only supports partitioning by a single column")
-
-            this = partitioned_by[0]
-
-            if (
-                isinstance(this, exp.Column)
-                and partition_interval_unit is not None
-                and not partition_interval_unit.is_minute
-            ):
-                column_type: t.Optional[exp.DataType] = (columns_to_types or {}).get(this.name)
-
-                if column_type == exp.DataType.build(
-                    "date", dialect=self.dialect
-                ) and partition_interval_unit in (
-                    IntervalUnit.MONTH,
-                    IntervalUnit.YEAR,
-                ):
-                    trunc_func = "DATE_TRUNC"
-                elif column_type == exp.DataType.build("timestamp", dialect=self.dialect):
-                    trunc_func = "TIMESTAMP_TRUNC"
-                elif column_type == exp.DataType.build("datetime", dialect=self.dialect):
-                    trunc_func = "DATETIME_TRUNC"
-                else:
-                    trunc_func = ""
-
-                if trunc_func:
-                    this = exp.func(
-                        trunc_func,
-                        this,
-                        exp.var(partition_interval_unit.value.upper()),
-                        dialect=self.dialect,
-                    )
-
-            properties.append(exp.PartitionedByProperty(this=this))
+        if partitioned_by and (
+            partitioned_by_prop := self._build_partitioned_by_exp(
+                partitioned_by,
+                partition_interval_unit=partition_interval_unit,
+                columns_to_types=columns_to_types,
+            )
+        ):
+            properties.append(partitioned_by_prop)
 
-        if clustered_by:
-            properties.append(exp.Cluster(expressions=[exp.column(col) for col in clustered_by]))
+        if clustered_by and (clustered_by_exp := self._build_clustered_by_exp(clustered_by)):
+            properties.append(clustered_by_exp)
 
         if table_description:
             properties.append(
diff --git a/sqlmesh/core/engine_adapter/clickhouse.py b/sqlmesh/core/engine_adapter/clickhouse.py
index a0473cf0..4d5df91b 100644
--- a/sqlmesh/core/engine_adapter/clickhouse.py
+++ b/sqlmesh/core/engine_adapter/clickhouse.py
@@ -287,6 +287,15 @@ class ClickhouseEngineAdapter(EngineAdapterWithIndexSupport, LogicalMergeMixin):
             )
         )
 
+    def _build_partitioned_by_exp(
+        self,
+        partitioned_by: t.List[exp.Expression],
+        **kwargs: t.Any,
+    ) -> t.Optional[t.Union[exp.PartitionedByProperty, exp.Property]]:
+        return exp.PartitionedByProperty(
+            this=exp.Schema(expressions=partitioned_by),
+        )
+
     def _build_table_properties_exp(
         self,
         catalog_name: t.Optional[str] = None,
@@ -364,12 +373,10 @@ class ClickhouseEngineAdapter(EngineAdapterWithIndexSupport, LogicalMergeMixin):
                 )
             )
 
-        if partitioned_by:
-            properties.append(
-                exp.PartitionedByProperty(
-                    this=exp.Schema(expressions=partitioned_by),
-                )
-            )
+        if partitioned_by and (
+            partitioned_by_prop := self._build_partitioned_by_exp(partitioned_by)
+        ):
+            properties.append(partitioned_by_prop)
 
         if self.engine_run_mode.is_cluster:
             properties.append(exp.OnCluster(this=exp.to_identifier(self.cluster)))
diff --git a/sqlmesh/core/engine_adapter/mixins.py b/sqlmesh/core/engine_adapter/mixins.py
index da896a22..da418506 100644
--- a/sqlmesh/core/engine_adapter/mixins.py
+++ b/sqlmesh/core/engine_adapter/mixins.py
@@ -116,6 +116,36 @@ class HiveMetastoreTablePropertiesMixin(EngineAdapter):
     MAX_TABLE_COMMENT_LENGTH = 4000
     MAX_COLUMN_COMMENT_LENGTH = 4000
 
+    def _build_partitioned_by_exp(
+        self,
+        partitioned_by: t.List[exp.Expression],
+        *,
+        catalog_name: t.Optional[str] = None,
+        **kwargs: t.Any,
+    ) -> t.Union[exp.PartitionedByProperty, exp.Property]:
+        if (
+            self.dialect == "trino"
+            and self.get_catalog_type(catalog_name or self.get_current_catalog()) == "iceberg"
+        ):
+            # On the Trino Iceberg catalog, the table property is called "partitioning" - not "partitioned_by"
+            # In addition, partition column transform expressions like `day(col)` or `bucket(col, 5)` are allowed
+            # Also, column names and transforms need to be strings and supplied as an ARRAY[varchar]
+            # ref: https://trino.io/docs/current/connector/iceberg.html#table-properties
+            return exp.Property(
+                this=exp.var("PARTITIONING"),
+                value=exp.array(
+                    *(exp.Literal.string(e.sql(dialect=self.dialect)) for e in partitioned_by)
+                ),
+            )
+        for expr in partitioned_by:
+            if not isinstance(expr, exp.Column):
+                raise SQLMeshError(
+                    f"PARTITIONED BY contains non-column value '{expr.sql(dialect=self.dialect)}'."
+                )
+        return exp.PartitionedByProperty(
+            this=exp.Schema(expressions=partitioned_by),
+        )
+
     def _build_table_properties_exp(
         self,
         catalog_name: t.Optional[str] = None,
@@ -134,37 +164,13 @@ class HiveMetastoreTablePropertiesMixin(EngineAdapter):
             properties.append(exp.FileFormatProperty(this=exp.Var(this=storage_format)))
 
         if partitioned_by:
-            if (
-                self.dialect == "trino"
-                and self.get_catalog_type(catalog_name or self.get_current_catalog()) == "iceberg"
-            ):
-                # On the Trino Iceberg catalog, the table property is called "partitioning" - not "partitioned_by"
-                # In addition, partition column transform expressions like `day(col)` or `bucket(col, 5)` are allowed
-                # Also, column names and transforms need to be strings and supplied as an ARRAY[varchar]
-                # ref: https://trino.io/docs/current/connector/iceberg.html#table-properties
-                properties.append(
-                    exp.Property(
-                        this=exp.var("PARTITIONING"),
-                        value=exp.array(
-                            *(
-                                exp.Literal.string(e.sql(dialect=self.dialect))
-                                for e in partitioned_by
-                            )
-                        ),
-                    )
-                )
-            else:
-                for expr in partitioned_by:
-                    if not isinstance(expr, exp.Column):
-                        raise SQLMeshError(
-                            f"PARTITIONED BY contains non-column value '{expr.sql(dialect=self.dialect)}'."
-                        )
-
-                properties.append(
-                    exp.PartitionedByProperty(
-                        this=exp.Schema(expressions=partitioned_by),
-                    )
+            properties.append(
+                self._build_partitioned_by_exp(
+                    partitioned_by,
+                    partition_interval_unit=partition_interval_unit,
+                    catalog_name=catalog_name,
                 )
+            )
 
         if table_description:
             properties.append(
@@ -313,3 +319,12 @@ class VarcharSizeWorkaroundMixin(EngineAdapter):
                 self.drop_view(temp_view_name)
 
         return statement
+
+
+class ClusteredByMixin(EngineAdapter):
+    def _build_clustered_by_exp(
+        self,
+        clustered_by: t.List[str],
+        **kwargs: t.Any,
+    ) -> t.Optional[exp.Cluster]:
+        return exp.Cluster(expressions=[exp.column(col) for col in clustered_by])
diff --git a/sqlmesh/core/engine_adapter/redshift.py b/sqlmesh/core/engine_adapter/redshift.py
index 77abb2bb..4f9a8009 100644
--- a/sqlmesh/core/engine_adapter/redshift.py
+++ b/sqlmesh/core/engine_adapter/redshift.py
@@ -127,6 +127,7 @@ class RedshiftEngineAdapter(
         columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
         replace: bool = True,
         materialized: bool = False,
+        materialized_properties: t.Optional[t.Dict[str, t.Any]] = None,
         table_description: t.Optional[str] = None,
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         view_properties: t.Optional[t.Dict[str, exp.Expression]] = None,
@@ -143,6 +144,7 @@ class RedshiftEngineAdapter(
             columns_to_types,
             replace,
             materialized,
+            materialized_properties,
             table_description=table_description,
             column_descriptions=column_descriptions,
             no_schema_binding=create_kwargs.pop("no_schema_binding", True),
diff --git a/sqlmesh/core/engine_adapter/snowflake.py b/sqlmesh/core/engine_adapter/snowflake.py
index 1121b0c9..0a79fe96 100644
--- a/sqlmesh/core/engine_adapter/snowflake.py
+++ b/sqlmesh/core/engine_adapter/snowflake.py
@@ -11,7 +11,7 @@ from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
 from sqlglot.optimizer.qualify_columns import quote_identifiers
 
 from sqlmesh.core.dialect import to_schema
-from sqlmesh.core.engine_adapter.mixins import GetCurrentCatalogFromFunctionMixin
+from sqlmesh.core.engine_adapter.mixins import GetCurrentCatalogFromFunctionMixin, ClusteredByMixin
 from sqlmesh.core.engine_adapter.shared import (
     CatalogSupport,
     DataObject,
@@ -39,7 +39,7 @@ if t.TYPE_CHECKING:
         "drop_schema": CatalogSupport.REQUIRES_SET_CATALOG,
     }
 )
-class SnowflakeEngineAdapter(GetCurrentCatalogFromFunctionMixin):
+class SnowflakeEngineAdapter(GetCurrentCatalogFromFunctionMixin, ClusteredByMixin):
     DIALECT = "snowflake"
     SUPPORTS_MATERIALIZED_VIEWS = True
     SUPPORTS_MATERIALIZED_VIEW_SCHEMA = True
@@ -178,8 +178,8 @@ class SnowflakeEngineAdapter(GetCurrentCatalogFromFunctionMixin):
                 )
             )
 
-        if clustered_by:
-            properties.append(exp.Cluster(expressions=[exp.column(col) for col in clustered_by]))
+        if clustered_by and (clustered_by_prop := self._build_clustered_by_exp(clustered_by)):
+            properties.append(clustered_by_prop)
 
         if table_properties:
             table_properties = {k.upper(): v for k, v in table_properties.items()}
diff --git a/sqlmesh/core/engine_adapter/spark.py b/sqlmesh/core/engine_adapter/spark.py
index b4fb6d41..101a2a25 100644
--- a/sqlmesh/core/engine_adapter/spark.py
+++ b/sqlmesh/core/engine_adapter/spark.py
@@ -376,6 +376,7 @@ class SparkEngineAdapter(GetCurrentCatalogFromFunctionMixin, HiveMetastoreTableP
         columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
         replace: bool = True,
         materialized: bool = False,
+        materialized_properties: t.Optional[t.Dict[str, t.Any]] = None,
         table_description: t.Optional[str] = None,
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         view_properties: t.Optional[t.Dict[str, exp.Expression]] = None,
@@ -392,9 +393,11 @@ class SparkEngineAdapter(GetCurrentCatalogFromFunctionMixin, HiveMetastoreTableP
             columns_to_types: Columns to use in the view statement.
             replace: Whether or not to replace an existing view - defaults to True.
             materialized: Whether or not the view should be materialized - defaults to False.
+            materialized_properties: Optional materialized view properties to add to the view.
             table_description: Optional table description from MODEL DDL.
             column_descriptions: Optional column descriptions from model query.
             create_kwargs: Additional kwargs to pass into the Create expression
+            view_properties: Optional view properties to add to the view.
         """
         pyspark_df = self.try_get_pyspark_df(query_or_df)
         if pyspark_df:
@@ -405,6 +408,7 @@ class SparkEngineAdapter(GetCurrentCatalogFromFunctionMixin, HiveMetastoreTableP
             columns_to_types,
             replace,
             materialized,
+            materialized_properties,
             table_description,
             column_descriptions,
             view_properties=view_properties,
diff --git a/sqlmesh/core/model/meta.py b/sqlmesh/core/model/meta.py
index 702a1463..3995bd2b 100644
--- a/sqlmesh/core/model/meta.py
+++ b/sqlmesh/core/model/meta.py
@@ -333,7 +333,11 @@ class ModelMeta(_Node):
         kind = values.get("kind")
         if kind:
             for field in ("partitioned_by_", "clustered_by"):
-                if values.get(field) and not kind.is_materialized:
+                if (
+                    values.get(field)
+                    and not kind.is_materialized
+                    and not (kind.is_view and kind.materialized)
+                ):
                     raise ValueError(f"{field} field cannot be set for {kind} models")
             if kind.is_incremental_by_partition and not values.get("partitioned_by_"):
                 raise ValueError(f"partitioned_by field is required for {kind.name} models")
diff --git a/sqlmesh/core/scheduler.py b/sqlmesh/core/scheduler.py
index c5e7cb19..6863b00e 100644
--- a/sqlmesh/core/scheduler.py
+++ b/sqlmesh/core/scheduler.py
@@ -26,7 +26,7 @@ from sqlmesh.core.snapshot.definition import Interval as SnapshotInterval
 from sqlmesh.core.snapshot.definition import SnapshotId
 from sqlmesh.core.state_sync import StateSync
 from sqlmesh.utils import format_exception
-from sqlmesh.utils.concurrency import concurrent_apply_to_dag
+from sqlmesh.utils.concurrency import concurrent_apply_to_dag, NodeExecutionFailedError
 from sqlmesh.utils.dag import DAG
 from sqlmesh.utils.date import (
     TimeLike,
@@ -347,14 +347,60 @@ class Scheduler:
         if not batches:
             return True
 
-        dag = self._dag(batches)
-
         self.console.start_evaluation_progress(
             {snapshot: len(intervals) for snapshot, intervals in batches.items()},
             environment_naming_info,
             self.default_catalog,
         )
 
+        errors, skipped_intervals = self.run_batches(
+            batches=batches,
+            deployability_index=deployability_index,
+            execution_time=execution_time,
+            circuit_breaker=circuit_breaker,
+        )
+
+        self.console.stop_evaluation_progress(success=not errors)
+
+        skipped_snapshots = {i[0] for i in skipped_intervals}
+        for skipped in skipped_snapshots:
+            log_message = f"SKIPPED snapshot {skipped}\n"
+            self.console.log_status_update(log_message)
+            logger.info(log_message)
+
+        for error in errors:
+            if isinstance(error.__cause__, CircuitBreakerError):
+                raise error.__cause__
+            sid = error.node[0]
+            formatted_exception = "".join(format_exception(error.__cause__ or error))
+            log_message = f"FAILED processing snapshot {sid}\n{formatted_exception}"
+            self.console.log_error(log_message)
+            # Log with INFO level to prevent duplicate messages in the console.
+            logger.info(log_message)
+
+        return not errors
+
+    def run_batches(
+        self,
+        batches: SnapshotToBatches,
+        deployability_index: DeployabilityIndex,
+        execution_time: t.Optional[TimeLike] = None,
+        circuit_breaker: t.Optional[t.Callable[[], bool]] = None,
+    ) -> t.Tuple[t.List[NodeExecutionFailedError[SchedulingUnit]], t.List[SchedulingUnit]]:
+        """Runs precomputed batches of missing intervals.
+
+        Args:
+            batches: The batches of snapshots and intervals to evaluate.
+            deployability_index: Determines snapshots that are deployable in the context of this evaluation.
+            execution_time: The date/time time reference to use for execution time.
+            circuit_breaker: An optional handler which checks if the run should be aborted.
+
+        Returns:
+            A tuple of errors and skipped intervals.
+        """
+        execution_time = execution_time or now()
+        dag = self._dag(batches)
+
         snapshots_by_name = {snapshot.name: snapshot for snapshot in self.snapshots.values()}
 
         def evaluate_node(node: SchedulingUnit) -> None:
@@ -383,7 +429,7 @@ class Scheduler:
 
         try:
             with self.snapshot_evaluator.concurrent_context():
-                errors, skipped_intervals = concurrent_apply_to_dag(
+                return concurrent_apply_to_dag(
                     dag,
                     evaluate_node,
                     self.max_workers,
@@ -392,26 +438,6 @@ class Scheduler:
         finally:
             self.state_sync.recycle()
 
-        self.console.stop_evaluation_progress(success=not errors)
-
-        skipped_snapshots = {i[0] for i in skipped_intervals}
-        for skipped in skipped_snapshots:
-            log_message = f"SKIPPED snapshot {skipped}\n"
-            self.console.log_status_update(log_message)
-            logger.info(log_message)
-
-        for error in errors:
-            if isinstance(error.__cause__, CircuitBreakerError):
-                raise error.__cause__
-            sid = error.node[0]
-            formatted_exception = "".join(format_exception(error.__cause__ or error))
-            log_message = f"FAILED processing snapshot {sid}\n{formatted_exception}"
-            self.console.log_error(log_message)
-            # Log with INFO level to prevent duplicate messages in the console.
-            logger.info(log_message)
-
-        return not errors
-
     def _dag(self, batches: SnapshotToBatches) -> DAG[SchedulingUnit]:
         """Builds a DAG of snapshot intervals to be evaluated.
 
diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py
index b72e6b3b..47439ce4 100644
--- a/sqlmesh/core/snapshot/evaluator.py
+++ b/sqlmesh/core/snapshot/evaluator.py
@@ -1593,12 +1593,21 @@ class ViewStrategy(PromotableStrategy):
             return
 
         logger.info("Creating view '%s'", table_name)
+        materialized = self._is_materialized_view(model)
+        materialized_properties = None
+        if materialized:
+            materialized_properties = {
+                "partitioned_by": model.partitioned_by,
+                "clustered_by": model.clustered_by,
+                "partition_interval_unit": model.interval_unit,
+            }
         self.adapter.create_view(
             table_name,
             model.render_query_or_raise(**render_kwargs),
             # Make sure we never replace the view during creation to avoid race conditions in engines with no late binding support.
             replace=False,
             materialized=self._is_materialized_view(model),
+            materialized_properties=materialized_properties,
             view_properties=model.physical_properties,
             table_description=model.description if is_table_deployable else None,
             column_descriptions=model.column_descriptions if is_table_deployable else None,


Test patch:
diff --git a/tests/core/engine_adapter/test_bigquery.py b/tests/core/engine_adapter/test_bigquery.py
index ad6e6fbe..c197ad08 100644
--- a/tests/core/engine_adapter/test_bigquery.py
+++ b/tests/core/engine_adapter/test_bigquery.py
@@ -755,3 +755,29 @@ def test_view_properties(make_mocked_engine_adapter: t.Callable, mocker: MockerF
         "CREATE OR REPLACE VIEW `test_table` OPTIONS (description='some description', labels=[('test-view-label', 'label-view-value')]) AS SELECT 1",
         "CREATE OR REPLACE VIEW `test_table` AS SELECT 1",
     ]
+
+
+def test_materialized_view_properties(
+    make_mocked_engine_adapter: t.Callable, mocker: MockerFixture
+):
+    adapter = make_mocked_engine_adapter(BigQueryEngineAdapter)
+    execute_mock = mocker.patch(
+        "sqlmesh.core.engine_adapter.bigquery.BigQueryEngineAdapter.execute"
+    )
+
+    adapter.create_view(
+        "test_table",
+        parse_one("SELECT 1"),
+        materialized=True,
+        materialized_properties={
+            "partitioned_by": [exp.column("ds")],
+            "clustered_by": ["a"],
+            "partition_interval_unit": IntervalUnit.DAY,
+        },
+    )
+
+    sql_calls = _to_sql_calls(execute_mock)
+    # https://cloud.google.com/bigquery/docs/materialized-views-create#example_1
+    assert sql_calls == [
+        "CREATE OR REPLACE MATERIALIZED VIEW `test_table` PARTITION BY `ds` CLUSTER BY `a` AS SELECT 1",
+    ]
diff --git a/tests/core/engine_adapter/test_databricks.py b/tests/core/engine_adapter/test_databricks.py
index a825af44..ada93f14 100644
--- a/tests/core/engine_adapter/test_databricks.py
+++ b/tests/core/engine_adapter/test_databricks.py
@@ -8,6 +8,7 @@ from sqlglot import exp, parse_one
 
 from sqlmesh.core import dialect as d
 from sqlmesh.core.engine_adapter import DatabricksEngineAdapter
+from sqlmesh.core.node import IntervalUnit
 from tests.core.engine_adapter import to_sql_calls
 
 pytestmark = [pytest.mark.databricks, pytest.mark.engine]
@@ -136,3 +137,25 @@ def test_insert_overwrite_by_partition_query(
         "INSERT INTO `test_schema`.`test_table` REPLACE WHERE CONCAT_WS('__SQLMESH_DELIM__', DATE_TRUNC('MONTH', `ds`), `b`) IN (SELECT DISTINCT CONCAT_WS('__SQLMESH_DELIM__', DATE_TRUNC('MONTH', `ds`), `b`) FROM `test_schema`.`temp_test_table_abcdefgh`) SELECT `a`, `ds`, `b` FROM `test_schema`.`temp_test_table_abcdefgh`",
         "DROP TABLE IF EXISTS `test_schema`.`temp_test_table_abcdefgh`",
     ]
+
+
+def test_materialized_view_properties(make_mocked_engine_adapter: t.Callable):
+    adapter = make_mocked_engine_adapter(DatabricksEngineAdapter)
+
+    adapter.create_view(
+        "test_table",
+        parse_one("SELECT 1"),
+        materialized=True,
+        materialized_properties={
+            "partitioned_by": [exp.column("ds")],
+            # Clustered by is not supported so we are confirming it is ignored
+            "clustered_by": ["a"],
+            "partition_interval_unit": IntervalUnit.DAY,
+        },
+    )
+
+    sql_calls = to_sql_calls(adapter)
+    # https://docs.databricks.com/en/sql/language-manual/sql-ref-syntax-ddl-create-materialized-view.html#syntax
+    assert sql_calls == [
+        "CREATE OR REPLACE MATERIALIZED VIEW test_table PARTITIONED BY (ds) AS SELECT 1",
+    ]
diff --git a/tests/core/engine_adapter/test_snowflake.py b/tests/core/engine_adapter/test_snowflake.py
index eb57cb3c..a1c9863a 100644
--- a/tests/core/engine_adapter/test_snowflake.py
+++ b/tests/core/engine_adapter/test_snowflake.py
@@ -10,6 +10,7 @@ from sqlmesh.core.dialect import normalize_model_name
 from sqlmesh.core.model import load_sql_based_model
 from sqlmesh.core.engine_adapter import SnowflakeEngineAdapter
 from sqlmesh.core.model.definition import SqlModel
+from sqlmesh.core.node import IntervalUnit
 from sqlmesh.utils.errors import SQLMeshError
 from sqlmesh.utils import optional_import
 from tests.core.engine_adapter import to_sql_calls
@@ -426,3 +427,25 @@ def test_replace_query_snowpark_dataframe(
         'CREATE OR REPLACE TABLE "foo" AS SELECT CAST("ID" AS INT) AS "ID", CAST("NAME" AS VARCHAR) AS "NAME" FROM (SELECT CAST("ID" AS INT) AS "ID", CAST("NAME" AS VARCHAR) AS "NAME" FROM "__temp_foo_e6wjkjj6") AS "_subquery"',
         'DROP VIEW IF EXISTS "__temp_foo_e6wjkjj6"',
     ]
+
+
+def test_materialized_view_properties(make_mocked_engine_adapter: t.Callable):
+    adapter = make_mocked_engine_adapter(SnowflakeEngineAdapter)
+
+    adapter.create_view(
+        "test_table",
+        parse_one("SELECT 1"),
+        materialized=True,
+        materialized_properties={
+            # Partitioned by is not supported so we are confirming it is ignored
+            "partitioned_by": [exp.column("ds")],
+            "clustered_by": ["a"],
+            "partition_interval_unit": IntervalUnit.DAY,
+        },
+    )
+
+    sql_calls = to_sql_calls(adapter)
+    # https://docs.snowflake.com/en/sql-reference/sql/create-materialized-view#syntax
+    assert sql_calls == [
+        'CREATE OR REPLACE MATERIALIZED VIEW "test_table" CLUSTER BY ("a") AS SELECT 1',
+    ]
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index f02fd46c..de953650 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -3381,6 +3381,55 @@ def test_view_model_data_hash():
     assert view_model_hash != materialized_view_model_hash
 
 
+def test_view_materialized_partition_by_clustered_by():
+    materialized_view_model_expressions = d.parse(
+        """
+        MODEL (
+            name db.table,
+            kind VIEW (
+              materialized true
+            ),
+            partitioned_by ds,
+            clustered_by a
+        );
+        SELECT 1;
+        """
+    )
+    materialized_view_model = load_sql_based_model(materialized_view_model_expressions)
+    assert materialized_view_model.partitioned_by == [exp.column("ds", quoted=True)]
+    assert materialized_view_model.clustered_by == ["a"]
+
+
+def test_view_non_materialized_partition_by():
+    view_model_expressions = d.parse(
+        """
+        MODEL (
+            name db.table,
+            kind VIEW,
+            partitioned_by ds,
+        );
+        SELECT 1;
+        """
+    )
+    with pytest.raises(ConfigError, match=r".*partitioned_by_ field cannot be set for ViewKind.*"):
+        load_sql_based_model(view_model_expressions)
+
+
+def test_view_non_materialized_clustered_by():
+    view_model_expressions = d.parse(
+        """
+        MODEL (
+            name db.table,
+            kind VIEW,
+            clustered_by ds,
+        );
+        SELECT 1;
+        """
+    )
+    with pytest.raises(ConfigError, match=r".*clustered_by field cannot be set for ViewKind.*"):
+        load_sql_based_model(view_model_expressions)
+
+
 def test_seed_model_data_hash():
     expressions = d.parse(
         """
diff --git a/tests/core/test_snapshot_evaluator.py b/tests/core/test_snapshot_evaluator.py
index e923f98a..c76151a4 100644
--- a/tests/core/test_snapshot_evaluator.py
+++ b/tests/core/test_snapshot_evaluator.py
@@ -13,7 +13,7 @@ from sqlglot import parse, parse_one, select
 from sqlmesh.core.audit import ModelAudit, StandaloneAudit
 from sqlmesh.core import dialect as d
 from sqlmesh.core.dialect import schema_, to_schema
-from sqlmesh.core.engine_adapter import EngineAdapter, create_engine_adapter
+from sqlmesh.core.engine_adapter import EngineAdapter, create_engine_adapter, BigQueryEngineAdapter
 from sqlmesh.core.engine_adapter.base import MERGE_SOURCE_ALIAS, MERGE_TARGET_ALIAS
 from sqlmesh.core.engine_adapter.shared import (
     DataObject,
@@ -511,6 +511,45 @@ def test_evaluate_materialized_view(
         )
 
 
+def test_evaluate_materialized_view_with_partitioned_by_cluster_by(
+    mocker: MockerFixture, adapter_mock, make_snapshot
+):
+    execute_mock = mocker.Mock()
+    # Use an engine adapter that supports cluster by/partitioned by
+    adapter = BigQueryEngineAdapter(lambda: mocker.Mock())
+    adapter.table_exists = lambda *args, **kwargs: False  # type: ignore
+    adapter.get_data_objects = lambda *args, **kwargs: []  # type: ignore
+    adapter._execute = execute_mock  # type: ignore
+    evaluator = SnapshotEvaluator(adapter)
+
+    model = SqlModel(
+        name="test_schema.test_model",
+        kind=ViewKind(
+            materialized=True,
+        ),
+        partitioned_by=[exp.to_column("a")],
+        clustered_by=[exp.to_column("b")],
+        query=parse_one("SELECT a, b FROM tbl"),
+    )
+
+    snapshot = make_snapshot(model)
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+    snapshot.add_interval("2023-01-01", "2023-01-01")
+
+    evaluator.create(
+        [snapshot],
+        snapshots={},
+    )
+
+    execute_mock.assert_has_calls(
+        [
+            call(
+                "CREATE MATERIALIZED VIEW `sqlmesh__test_schema`.`test_schema__test_model__2383078413` PARTITION BY `a` CLUSTER BY `b` AS SELECT `a` AS `a`, `b` AS `b` FROM `tbl` AS `tbl`"
+            ),
+        ]
+    )
+
+
 def test_evaluate_materialized_view_with_execution_time_macro(
     mocker: MockerFixture, adapter_mock, make_snapshot
 ):
@@ -741,6 +780,7 @@ def test_create_view_non_deployable_snapshot(mocker: MockerFixture, adapter_mock
         table_description=None,
         materialized=False,
         replace=False,
+        materialized_properties=None,
     )
 
 
@@ -772,6 +812,11 @@ def test_create_materialized_view(mocker: MockerFixture, adapter_mock, make_snap
 
     common_kwargs = dict(
         materialized=True,
+        materialized_properties={
+            "clustered_by": [],
+            "partition_interval_unit": IntervalUnit.DAY,
+            "partitioned_by": [],
+        },
         view_properties={},
         table_description=None,
         replace=False,
@@ -826,6 +871,11 @@ def test_create_view_with_properties(mocker: MockerFixture, adapter_mock, make_s
         view_properties={
             "key": exp.convert("value"),
         },
+        materialized_properties={
+            "clustered_by": [],
+            "partition_interval_unit": IntervalUnit.DAY,
+            "partitioned_by": [],
+        },
         table_description=None,
         replace=False,
     )