Repository: TobikoData/sqlmesh
Pull request #3192 (2024-09-27 00:13:32)

fix!: correctly use execution time on new rows for scd type 2 by column
This corrects the SCD Type 2 By Column default behavior to use execution time for new rows after the table is initially loaded. This matches what the documentation example says it should behave. This does mean there is a behavior change for current users but the current behavior seems to be a bug. There is not a way for a user to maintain old behavior if they wanted it since we consider it a bug. 

From reviewing the code, `truncate` is a great way to identify if a table is being loaded for the first time. This also means that if a user issues a restatement on the table and wants to rebuild from scratch then it would also be set to `truncate` and would mimic loading the rows for the first time. 


Patch:
diff --git a/docs/concepts/models/model_kinds.md b/docs/concepts/models/model_kinds.md
index 8b015687..dea7a5c4 100644
--- a/docs/concepts/models/model_kinds.md
+++ b/docs/concepts/models/model_kinds.md
@@ -736,10 +736,10 @@ This is the most accurate representation of the menu based on the source data pr
 
 ### SCD Type 2 By Column Configuration Options
 
-| Name                         | Description                                                                                                                                                                   | Type                      |
-|------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------------------|
-| columns                      | The name of the columns to check for changes. `*` to represent that all columns should be checked.                                                                            | List of strings or string |
-| 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                      |
+| Name                         | Description                                                                                                                                                                                                                                 | Type                      |
+|------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------------------|
+| columns                      | The name of the columns to check for changes. `*` to represent that all columns should be checked.                                                                                                                                          | List of strings or string |
+| execution_time_as_valid_from | By default, when the model is first loaded `valid_from` is set to `1970-01-01 00:00:00` and future new rows will have `execution_time` of when the pipeline ran. This changes the behavior to always use `execution_time`. Default: `false` | bool                      |
 
 ### Querying SCD Type 2 Models
 
diff --git a/sqlmesh/core/config/connection.py b/sqlmesh/core/config/connection.py
index a943f459..ed033b35 100644
--- a/sqlmesh/core/config/connection.py
+++ b/sqlmesh/core/config/connection.py
@@ -1432,6 +1432,12 @@ class ClickhouseConnectionConfig(ConnectionConfig):
     register_comments: bool = True
     pre_ping: bool = False
 
+    # This object expects options from urllib3 and also from clickhouse-connect
+    # See:
+    # * https://urllib3.readthedocs.io/en/stable/advanced-usage.html
+    # * https://clickhouse.com/docs/en/integrations/python#customizing-the-http-connection-pool
+    connection_pool_options: t.Optional[t.Dict[str, t.Any]] = None
+
     type_: Literal["clickhouse"] = Field(alias="type", default="clickhouse")
 
     @property
@@ -1455,8 +1461,20 @@ class ClickhouseConnectionConfig(ConnectionConfig):
     @property
     def _connection_factory(self) -> t.Callable:
         from clickhouse_connect.dbapi import connect  # type: ignore
+        from clickhouse_connect.driver import httputil  # type: ignore
+        from functools import partial
+
+        pool_manager_options: t.Dict[str, t.Any] = dict(
+            # Match the maxsize to the number of concurrent tasks
+            maxsize=self.concurrent_tasks,
+            # Block if there are no free connections
+            block=True,
+        )
+        if self.connection_pool_options:
+            pool_manager_options.update(self.connection_pool_options)
+        pool_mgr = httputil.get_pool_manager(**pool_manager_options)
 
-        return connect
+        return partial(connect, pool_mgr=pool_mgr)
 
     @property
     def cloud_mode(self) -> bool:
@@ -1497,7 +1515,11 @@ class ClickhouseConnectionConfig(ConnectionConfig):
             #   - "use majority number (number_of_replicas / 2 + 1) as quorum number"
             settings["insert_quorum"] = "auto"
 
-        return {"compress": compress, "client_name": f"SQLMesh/{__version__}", **settings}
+        return {
+            "compress": compress,
+            "client_name": f"SQLMesh/{__version__}",
+            **settings,
+        }
 
 
 class AthenaConnectionConfig(ConnectionConfig):
@@ -1575,7 +1597,9 @@ CONNECTION_CONFIG_TO_TYPE = {
     # Map all subclasses of ConnectionConfig to the value of their `type_` field.
     tpe.all_field_infos()["type_"].default: tpe
     for tpe in subclasses(
-        __name__, ConnectionConfig, exclude=(ConnectionConfig, BaseDuckDBConnectionConfig)
+        __name__,
+        ConnectionConfig,
+        exclude=(ConnectionConfig, BaseDuckDBConnectionConfig),
     )
 }
 
diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py
index 3fcf9b4b..4c669b75 100644
--- a/sqlmesh/core/context.py
+++ b/sqlmesh/core/context.py
@@ -453,8 +453,9 @@ class GenericContext(BaseContext, t.Generic[C]):
 
         update_model_schemas(
             self.dag,
-            self._models,
-            self.path,
+            models=self._models,
+            audits=self._audits,
+            context_path=self.path,
         )
 
         if model.dialect:
@@ -2057,7 +2058,8 @@ class GenericContext(BaseContext, t.Generic[C]):
     def _new_selector(self) -> Selector:
         return Selector(
             self.state_reader,
-            self._models,
+            models=self._models,
+            audits=self._audits,
             context_path=self.path,
             default_catalog=self.default_catalog,
             dialect=self.default_dialect,
diff --git a/sqlmesh/core/engine_adapter/base.py b/sqlmesh/core/engine_adapter/base.py
index e2be37ad..251f501c 100644
--- a/sqlmesh/core/engine_adapter/base.py
+++ b/sqlmesh/core/engine_adapter/base.py
@@ -1026,7 +1026,7 @@ class EngineAdapter:
             )
             and self.comments_enabled
         ):
-            self._create_column_comments(view_name, column_descriptions, "VIEW")
+            self._create_column_comments(view_name, column_descriptions, "VIEW", materialized)
 
     @set_catalog()
     def create_schema(
@@ -1453,7 +1453,10 @@ class EngineAdapter:
                     "Cannot use `updated_at_as_valid_from` without `updated_at_name` for SCD Type 2"
                 )
             update_valid_from_start: t.Union[str, exp.Expression] = updated_at_col
-        elif execution_time_as_valid_from:
+        # If using check_columns and the user doesn't always want execution_time for valid from
+        # then we only use epoch 0 if we are truncating the table and loading rows for the first time.
+        # All future new rows should have execution time.
+        elif check_columns and (execution_time_as_valid_from or not truncate):
             update_valid_from_start = execution_ts
         else:
             update_valid_from_start = to_time_column("1970-01-01 00:00:00+00:00", time_data_type)
@@ -1631,7 +1634,7 @@ class EngineAdapter:
                     .group_by(*unique_key),
                 )
                 # Do a full join between latest records and source table in order to combine them together
-                # MySQL doesn't suport full join so going to do a left then right join and remove dups with union
+                # MySQL doesn't support full join so going to do a left then right join and remove dups with union
                 # We do a left/right and filter right on only matching to remove the need to do union distinct
                 # which allows scd type 2 to be compatible with unhashable data types
                 .with_(
@@ -2293,6 +2296,7 @@ class EngineAdapter:
         table_name: TableName,
         column_comments: t.Dict[str, str],
         table_kind: str = "TABLE",
+        materialized_view: bool = False,
     ) -> None:
         table = exp.to_table(table_name)
 
diff --git a/sqlmesh/core/engine_adapter/bigquery.py b/sqlmesh/core/engine_adapter/bigquery.py
index cdad08f1..08f3cc2b 100644
--- a/sqlmesh/core/engine_adapter/bigquery.py
+++ b/sqlmesh/core/engine_adapter/bigquery.py
@@ -547,31 +547,33 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin, ClusteredByMixin):
         table_name: TableName,
         column_comments: t.Dict[str, str],
         table_kind: str = "TABLE",
+        materialized_view: bool = False,
     ) -> None:
-        table = self._get_table(table_name)
-
-        # convert Table object to dict
-        table_def = table.to_api_repr()
-
-        # set the column descriptions
-        for i in range(len(table_def["schema"]["fields"])):
-            comment = column_comments.get(table_def["schema"]["fields"][i]["name"], None)
-            if comment:
-                table_def["schema"]["fields"][i]["description"] = self._truncate_comment(
-                    comment, self.MAX_COLUMN_COMMENT_LENGTH
-                )
+        if not table_kind == "VIEW" and materialized_view:
+            table = self._get_table(table_name)
+
+            # convert Table object to dict
+            table_def = table.to_api_repr()
+
+            # set the column descriptions
+            for i in range(len(table_def["schema"]["fields"])):
+                comment = column_comments.get(table_def["schema"]["fields"][i]["name"], None)
+                if comment:
+                    table_def["schema"]["fields"][i]["description"] = self._truncate_comment(
+                        comment, self.MAX_COLUMN_COMMENT_LENGTH
+                    )
 
-        # An "etag" is BQ versioning metadata that changes when an object is updated/modified. `update_table`
-        # compares the etags of the table object passed to it and the remote table, erroring if the etags
-        # don't match. We set the local etag to None to avoid this check.
-        table_def["etag"] = None
+            # An "etag" is BQ versioning metadata that changes when an object is updated/modified. `update_table`
+            # compares the etags of the table object passed to it and the remote table, erroring if the etags
+            # don't match. We set the local etag to None to avoid this check.
+            table_def["etag"] = None
 
-        # convert dict back to a Table object
-        table = table.from_api_repr(table_def)
+            # convert dict back to a Table object
+            table = table.from_api_repr(table_def)
 
-        # update table schema
-        logger.info(f"Registering column comments for table {table_name}")
-        self._db_call(self.client.update_table, table=table, fields=["schema"])
+            # update table schema
+            logger.info(f"Registering column comments for table {table_name}")
+            self._db_call(self.client.update_table, table=table, fields=["schema"])
 
     def _build_description_property_exp(
         self,
diff --git a/sqlmesh/core/engine_adapter/mysql.py b/sqlmesh/core/engine_adapter/mysql.py
index ad244b79..7f208a32 100644
--- a/sqlmesh/core/engine_adapter/mysql.py
+++ b/sqlmesh/core/engine_adapter/mysql.py
@@ -130,6 +130,7 @@ class MySQLEngineAdapter(
         table_name: TableName,
         column_comments: t.Dict[str, str],
         table_kind: str = "TABLE",
+        materialized_view: bool = False,
     ) -> None:
         table = exp.to_table(table_name)
         table_sql = table.sql(dialect=self.dialect, identify=True)
diff --git a/sqlmesh/core/engine_adapter/snowflake.py b/sqlmesh/core/engine_adapter/snowflake.py
index d7f2bdb8..047c310a 100644
--- a/sqlmesh/core/engine_adapter/snowflake.py
+++ b/sqlmesh/core/engine_adapter/snowflake.py
@@ -394,6 +394,7 @@ class SnowflakeEngineAdapter(GetCurrentCatalogFromFunctionMixin, ClusteredByMixi
         table_name: TableName,
         column_comments: t.Dict[str, str],
         table_kind: str = "TABLE",
+        materialized_view: bool = False,
     ) -> None:
         """
         Reference: https://docs.snowflake.com/en/sql-reference/sql/alter-table-column#syntax
diff --git a/sqlmesh/core/loader.py b/sqlmesh/core/loader.py
index 609c7ea8..e6977e6f 100644
--- a/sqlmesh/core/loader.py
+++ b/sqlmesh/core/loader.py
@@ -14,7 +14,7 @@ from sqlglot.errors import SchemaError, SqlglotError
 from sqlglot.schema import MappingSchema
 
 from sqlmesh.core import constants as c
-from sqlmesh.core.audit import Audit, load_multiple_audits
+from sqlmesh.core.audit import Audit, ModelAudit, load_multiple_audits
 from sqlmesh.core.dialect import parse
 from sqlmesh.core.macros import MacroRegistry, macro
 from sqlmesh.core.metric import Metric, MetricMeta, expand_metrics, load_metric_ddl
@@ -27,7 +27,7 @@ from sqlmesh.core.model import (
     create_external_model,
     load_sql_based_model,
 )
-from sqlmesh.core.model.cache import optimized_query_cache_pool, load_optimized_query_cache
+from sqlmesh.core.model.cache import load_optimized_query_and_mapping, optimized_query_cache_pool
 from sqlmesh.core.model import model as model_registry
 from sqlmesh.utils import UniqueKeyDict
 from sqlmesh.utils.dag import DAG
@@ -104,8 +104,9 @@ class Loader(abc.ABC):
         if update_schemas:
             update_model_schemas(
                 self._dag,
-                models,
-                self._context.path,
+                models=models,
+                audits={k: v for k, v in audits.items() if isinstance(v, ModelAudit)},
+                context_path=self._context.path,
             )
             for model in models.values():
                 # The model definition can be validated correctly only after the schema is set.
@@ -552,6 +553,7 @@ class SqlMeshLoader(Loader):
 def update_model_schemas(
     dag: DAG[str],
     models: UniqueKeyDict[str, Model],
+    audits: t.Dict[str, ModelAudit],
     context_path: Path,
 ) -> None:
     schema = MappingSchema(normalize=False)
@@ -560,14 +562,14 @@ def update_model_schemas(
     if c.MAX_FORK_WORKERS == 1:
         _update_model_schemas_sequential(dag, models, schema, optimized_query_cache)
     else:
-        _update_model_schemas_parallel(dag, models, schema, optimized_query_cache)
+        _update_model_schemas_parallel(dag, models, audits, schema, optimized_query_cache)
 
 
 def _update_schema_with_model(schema: MappingSchema, model: Model) -> None:
     columns_to_types = model.columns_to_types
     if columns_to_types:
         try:
-            schema.add_table(model.fqn, columns_to_types, dialect=model.dialect, normalize=False)
+            schema.add_table(model.fqn, columns_to_types, dialect=model.dialect)
         except SchemaError as e:
             if "nesting level:" in str(e):
                 logger.error(
@@ -597,6 +599,7 @@ def _update_model_schemas_sequential(
 def _update_model_schemas_parallel(
     dag: DAG[str],
     models: UniqueKeyDict[str, Model],
+    audits: t.Dict[str, ModelAudit],
     schema: MappingSchema,
     optimized_query_cache: OptimizedQueryCache,
 ) -> None:
@@ -617,19 +620,29 @@ def _update_model_schemas_parallel(
             if not deps:
                 del graph[name]
                 model = models[name]
-                model.update_schema(schema)
-                futures.add(executor.submit(load_optimized_query_cache, model))
+                futures.add(
+                    executor.submit(
+                        load_optimized_query_and_mapping,
+                        model,
+                        mapping={
+                            parent: models[parent].columns_to_types
+                            for parent in model.depends_on
+                            if parent in models
+                        },
+                    )
+                )
 
-    with optimized_query_cache_pool(optimized_query_cache) as executor:
+    with optimized_query_cache_pool(optimized_query_cache, audits) as executor:
         process_models()
 
         while futures:
             for future in as_completed(futures):
                 futures.remove(future)
-                fqn, entry_name = future.result()
+                fqn, entry_name, data_hash, metadata_hash, mapping_schema = future.result()
                 model = models[fqn]
-                if entry_name:
-                    optimized_query_cache.with_optimized_query(model, entry_name)
-
+                model._data_hash = data_hash
+                model._metadata_hash = metadata_hash
+                model.set_mapping_schema(mapping_schema)
+                optimized_query_cache.with_optimized_query(model, entry_name)
                 _update_schema_with_model(schema, model)
                 process_models(completed_model=model)
diff --git a/sqlmesh/core/model/cache.py b/sqlmesh/core/model/cache.py
index 4a79634a..7b31a988 100644
--- a/sqlmesh/core/model/cache.py
+++ b/sqlmesh/core/model/cache.py
@@ -8,6 +8,7 @@ from pathlib import Path
 
 from sqlglot import exp
 from sqlglot.optimizer.simplify import gen
+from sqlglot.schema import MappingSchema
 
 from sqlmesh.core import constants as c
 from sqlmesh.core.model.definition import Model, SqlModel, _Model
@@ -18,7 +19,11 @@ from dataclasses import dataclass
 
 logger = logging.getLogger(__name__)
 
-T = t.TypeVar("T")
+if t.TYPE_CHECKING:
+    from sqlmesh.core.audit import ModelAudit
+    from sqlmesh.core.snapshot import SnapshotId
+
+    T = t.TypeVar("T")
 
 
 class ModelCache:
@@ -133,47 +138,68 @@ class OptimizedQueryCache:
         return f"{model.name}_{crc32(hash_data)}"
 
 
-def optimized_query_cache_pool(optimized_query_cache: OptimizedQueryCache) -> ProcessPoolExecutor:
+def optimized_query_cache_pool(
+    optimized_query_cache: OptimizedQueryCache,
+    audits: t.Optional[t.Dict[str, ModelAudit]] = None,
+) -> ProcessPoolExecutor:
     return ProcessPoolExecutor(
         mp_context=mp.get_context("fork"),
         initializer=_init_optimized_query_cache,
-        initargs=(optimized_query_cache,),
+        initargs=(optimized_query_cache, audits),
         max_workers=c.MAX_FORK_WORKERS,
     )
 
 
-@t.overload
-def load_optimized_query_cache(
-    model_or_tuple: t.Tuple[Model, T],
-) -> t.Tuple[T, t.Optional[str]]: ...
+_optimized_query_cache: t.Optional[OptimizedQueryCache] = None
+_audits: t.Optional[t.Dict[str, ModelAudit]] = None
 
 
-@t.overload
-def load_optimized_query_cache(model_or_tuple: Model) -> t.Tuple[str, t.Optional[str]]: ...
+def _init_optimized_query_cache(
+    optimized_query_cache: OptimizedQueryCache,
+    audits: t.Optional[t.Dict[str, ModelAudit]] = None,
+) -> None:
+    global _optimized_query_cache, _audits
+    _optimized_query_cache = optimized_query_cache
+    _audits = audits
 
 
-def load_optimized_query_cache(model_or_tuple):  # type: ignore
+def load_optimized_query(
+    model_snapshot_id: t.Tuple[Model, SnapshotId],
+) -> t.Tuple[SnapshotId, t.Optional[str]]:
     assert _optimized_query_cache
-
-    if isinstance(model_or_tuple, _Model):
-        model = model_or_tuple
-        key = None
-    else:
-        model, key = model_or_tuple
+    model, snapshot_id = model_snapshot_id
 
     if isinstance(model, SqlModel):
         entry_name = _optimized_query_cache.put(model)
     else:
         entry_name = None
-    return key or model.fqn, entry_name
+    return snapshot_id, entry_name
 
 
-_optimized_query_cache: t.Optional[OptimizedQueryCache] = None
+def load_optimized_query_and_mapping(
+    model: Model, mapping: t.Dict
+) -> t.Tuple[str, t.Optional[str], str, str, t.Dict]:
+    assert _optimized_query_cache
+    assert _audits is not None
 
+    schema = MappingSchema(normalize=False)
+    for parent, columns_to_types in mapping.items():
+        schema.add_table(parent, columns_to_types, dialect=model.dialect)
+    model.update_schema(schema)
 
-def _init_optimized_query_cache(optimized_query_cache: OptimizedQueryCache) -> None:
-    global _optimized_query_cache
-    _optimized_query_cache = optimized_query_cache
+    if isinstance(model, SqlModel):
+        entry_name = _optimized_query_cache._entry_name(model)
+        _optimized_query_cache.with_optimized_query(model, entry_name)
+    else:
+        entry_name = None
+
+    return (
+        model.fqn,
+        entry_name,
+        model.data_hash,
+        model.metadata_hash(_audits),
+        model.mapping_schema,
+    )
 
 
 def _mapping_schema_hash_data(schema: t.Dict[str, t.Any]) -> t.List[str]:
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index 38ba2939..592bb63f 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -627,10 +627,11 @@ class _Model(ModelMeta, frozen=True):
             return to_time_column(time, time_column_type, self.time_column.format)
         return exp.convert(time)
 
-    def update_schema(
-        self,
-        schema: MappingSchema,
-    ) -> None:
+    def set_mapping_schema(self, schema: t.Dict) -> None:
+        self.mapping_schema.clear()
+        self.mapping_schema.update(schema)
+
+    def update_schema(self, schema: MappingSchema) -> None:
         """Updates the schema for this model's dependencies based on the given mapping schema."""
         for dep in self.depends_on:
             table = exp.to_table(dep)
@@ -1109,11 +1110,15 @@ class SqlModel(_SqlBasedModel):
             if select.comments
         }
 
-    def update_schema(
-        self,
-        schema: MappingSchema,
-    ) -> None:
+    def set_mapping_schema(self, schema: t.Dict) -> None:
+        super().set_mapping_schema(schema)
+        self._on_mapping_schema_set()
+
+    def update_schema(self, schema: MappingSchema) -> None:
         super().update_schema(schema)
+        self._on_mapping_schema_set()
+
+    def _on_mapping_schema_set(self) -> None:
         self._columns_to_types = None
         self._query_renderer.update_schema(self.mapping_schema)
 
diff --git a/sqlmesh/core/selector.py b/sqlmesh/core/selector.py
index 7176c7cf..a97bb660 100644
--- a/sqlmesh/core/selector.py
+++ b/sqlmesh/core/selector.py
@@ -11,8 +11,6 @@ from sqlglot import exp
 from sqlmesh.core.dialect import normalize_model_name
 from sqlmesh.core.environment import Environment
 from sqlmesh.core.loader import update_model_schemas
-from sqlmesh.core.model import Model
-from sqlmesh.core.state_sync import StateReader
 from sqlmesh.utils import UniqueKeyDict
 from sqlmesh.utils.dag import DAG
 from sqlmesh.utils.git import GitClient
@@ -20,11 +18,18 @@ from sqlmesh.utils.git import GitClient
 logger = logging.getLogger(__name__)
 
 
+if t.TYPE_CHECKING:
+    from sqlmesh.core.audit import ModelAudit
+    from sqlmesh.core.model import Model
+    from sqlmesh.core.state_sync import StateReader
+
+
 class Selector:
     def __init__(
         self,
         state_reader: StateReader,
         models: UniqueKeyDict[str, Model],
+        audits: t.Dict[str, ModelAudit],
         context_path: Path = Path("."),
         dag: t.Optional[DAG[str]] = None,
         default_catalog: t.Optional[str] = None,
@@ -32,6 +37,7 @@ class Selector:
     ):
         self._state_reader = state_reader
         self._models = models
+        self._audits = audits
         self._context_path = context_path
         self._default_catalog = default_catalog
         self._dialect = dialect
@@ -144,7 +150,9 @@ class Selector:
             models[model.fqn] = model
 
         if needs_update:
-            update_model_schemas(dag, models, self._context_path)
+            update_model_schemas(
+                dag, models=models, audits=self._audits, context_path=self._context_path
+            )
 
         return models
 
diff --git a/sqlmesh/core/snapshot/cache.py b/sqlmesh/core/snapshot/cache.py
index 0ace6f14..f99dca64 100644
--- a/sqlmesh/core/snapshot/cache.py
+++ b/sqlmesh/core/snapshot/cache.py
@@ -6,7 +6,7 @@ from pathlib import Path
 from sqlmesh.core.model.cache import (
     OptimizedQueryCache,
     optimized_query_cache_pool,
-    load_optimized_query_cache,
+    load_optimized_query,
 )
 from sqlmesh.core import constants as c
 from sqlmesh.core.snapshot.definition import Snapshot, SnapshotId
@@ -54,7 +54,7 @@ class SnapshotCache:
         if c.MAX_FORK_WORKERS != 1:
             with optimized_query_cache_pool(self._optimized_query_cache) as executor:
                 for key, entry_name in executor.map(
-                    load_optimized_query_cache,
+                    load_optimized_query,
                     (
                         (snapshot.model, s_id)
                         for s_id, snapshot in snapshots.items()


Test patch:
diff --git a/tests/core/engine_adapter/integration/test_integration.py b/tests/core/engine_adapter/integration/test_integration.py
index 119b2382..81698c91 100644
--- a/tests/core/engine_adapter/integration/test_integration.py
+++ b/tests/core/engine_adapter/integration/test_integration.py
@@ -869,6 +869,7 @@ def test_scd_type_2_by_time(ctx: TestContext):
         updated_at_as_valid_from=False,
         columns_to_types=input_schema,
         table_format=table_format,
+        truncate=True,
     )
     results = ctx.get_metadata_results()
     assert len(results.views) == 0
@@ -930,6 +931,7 @@ def test_scd_type_2_by_time(ctx: TestContext):
         updated_at_as_valid_from=False,
         columns_to_types=input_schema,
         table_format=table_format,
+        truncate=False,
     )
     results = ctx.get_metadata_results()
     assert len(results.views) == 0
@@ -1018,6 +1020,7 @@ def test_scd_type_2_by_column(ctx: TestContext):
         execution_time="2023-01-01",
         execution_time_as_valid_from=False,
         columns_to_types=ctx.columns_to_types,
+        truncate=True,
     )
     results = ctx.get_metadata_results()
     assert len(results.views) == 0
@@ -1087,6 +1090,7 @@ def test_scd_type_2_by_column(ctx: TestContext):
         execution_time="2023-01-05 00:00:00",
         execution_time_as_valid_from=False,
         columns_to_types=ctx.columns_to_types,
+        truncate=False,
     )
     results = ctx.get_metadata_results()
     assert len(results.views) == 0
@@ -1143,7 +1147,7 @@ def test_scd_type_2_by_column(ctx: TestContext):
                     "id": 5,
                     "name": "e",
                     "status": "inactive",
-                    "valid_from": "1970-01-01 00:00:00",
+                    "valid_from": "2023-01-05 00:00:00",
                     "valid_to": pd.NaT,
                 },
             ]
diff --git a/tests/core/engine_adapter/test_base.py b/tests/core/engine_adapter/test_base.py
index 5c99c395..bcb14ebd 100644
--- a/tests/core/engine_adapter/test_base.py
+++ b/tests/core/engine_adapter/test_base.py
@@ -1756,7 +1756,7 @@ WITH "source" AS (
     COALESCE("joined"."t_id", "joined"."id") AS "id",
     COALESCE("joined"."t_name", "joined"."name") AS "name",
     COALESCE("joined"."t_price", "joined"."price") AS "price",
-    COALESCE("t_test_VALID_from", CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AS "test_VALID_from",
+    COALESCE("t_test_VALID_from", CAST('2020-01-01 00:00:00' AS TIMESTAMP)) AS "test_VALID_from",
     CASE
       WHEN "joined"."_exists" IS NULL
       OR (
@@ -1947,7 +1947,7 @@ WITH "source" AS (
     COALESCE("joined"."t_id_b", "joined"."id_b") AS "id_b",
     COALESCE("joined"."t_name", "joined"."name") AS "name",
     COALESCE("joined"."t_price", "joined"."price") AS "price",
-    COALESCE("t_test_VALID_from", CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AS "test_VALID_from",
+    COALESCE("t_test_VALID_from", CAST('2020-01-01 00:00:00' AS TIMESTAMP)) AS "test_VALID_from",
     CASE
       WHEN "joined"."_exists" IS NULL
       OR (
@@ -2311,7 +2311,7 @@ WITH "source" AS (
     COALESCE("joined"."t_id", "joined"."id") AS "id",
     COALESCE("joined"."t_name", "joined"."name") AS "name",
     COALESCE("joined"."t_price", "joined"."price") AS "price",
-    COALESCE("t_test_valid_from", CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AS "test_valid_from",
+    COALESCE("t_test_valid_from", CAST('2020-01-01 00:00:00' AS TIMESTAMP)) AS "test_valid_from",
     CASE
       WHEN "joined"."_exists" IS NULL
       OR (
@@ -2506,7 +2506,7 @@ WITH "source" AS (
     COALESCE("joined"."t_id", "joined"."id") AS "id",
     COALESCE("joined"."t_name", "joined"."name") AS "name",
     COALESCE("joined"."t_price", "joined"."price") AS "price",
-    COALESCE("t_test_valid_from", CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AS "test_valid_from",
+    COALESCE("t_test_valid_from", CAST('2020-01-01 00:00:00' AS TIMESTAMP)) AS "test_valid_from",
     CASE
       WHEN (
         NOT "joined"."t_id" IS NULL AND NOT "joined"."id" IS NULL
diff --git a/tests/core/engine_adapter/test_bigquery.py b/tests/core/engine_adapter/test_bigquery.py
index c197ad08..2ef36578 100644
--- a/tests/core/engine_adapter/test_bigquery.py
+++ b/tests/core/engine_adapter/test_bigquery.py
@@ -623,24 +623,39 @@ def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture)
         column_descriptions={"a": long_column_comment},
     )
 
+    adapter._create_table_comment(
+        "test_table",
+        long_table_comment,
+    )
+
     adapter.create_view(
         "test_table",
         parse_one("SELECT a, b FROM source_table"),
         table_description=long_table_comment,
     )
 
-    adapter._create_table_comment(
+    # Bigquery doesn't support column comments for materialized views
+    db_call_mock = mocker.patch(
+        "sqlmesh.core.engine_adapter.bigquery.BigQueryEngineAdapter._db_call"
+    )
+
+    adapter.create_view(
         "test_table",
-        long_table_comment,
+        parse_one("SELECT a, b FROM source_table"),
+        table_description=long_table_comment,
+        column_descriptions={"a": long_column_comment},
+        materialized=True,
     )
+    assert not db_call_mock.called
 
     sql_calls = _to_sql_calls(execute_mock)
     assert sql_calls == [
         f"CREATE TABLE IF NOT EXISTS `test_table` (`a` INT64 OPTIONS (description='{truncated_column_comment}'), `b` INT64) OPTIONS (description='{truncated_table_comment}')",
         "CREATE TABLE IF NOT EXISTS `test_table` (`a` INT64 OPTIONS (description='\\\\'), `b` INT64) OPTIONS (description='\\\\')",
         f"CREATE TABLE IF NOT EXISTS `test_table` (`a` INT64 OPTIONS (description='{truncated_column_comment}'), `b` INT64) OPTIONS (description='{truncated_table_comment}') AS SELECT CAST(`a` AS INT64) AS `a`, CAST(`b` AS INT64) AS `b` FROM (SELECT `a`, `b` FROM `source_table`) AS `_subquery`",
-        f"CREATE OR REPLACE VIEW `test_table` OPTIONS (description='{truncated_table_comment}') AS SELECT `a`, `b` FROM `source_table`",
         f"ALTER TABLE `test_table` SET OPTIONS(description = '{truncated_table_comment}')",
+        f"CREATE OR REPLACE VIEW `test_table` OPTIONS (description='{truncated_table_comment}') AS SELECT `a`, `b` FROM `source_table`",
+        f"CREATE OR REPLACE MATERIALIZED VIEW `test_table` OPTIONS (description='{truncated_table_comment}') AS SELECT `a`, `b` FROM `source_table`",
     ]
 
 
diff --git a/tests/core/engine_adapter/test_clickhouse.py b/tests/core/engine_adapter/test_clickhouse.py
index e66bab5f..055b70ba 100644
--- a/tests/core/engine_adapter/test_clickhouse.py
+++ b/tests/core/engine_adapter/test_clickhouse.py
@@ -745,7 +745,7 @@ WITH "source" AS (
     COALESCE("joined"."t_id", "joined"."id") AS "id",
     COALESCE("joined"."t_name", "joined"."name") AS "name",
     COALESCE("joined"."t_price", "joined"."price") AS "price",
-    COALESCE("t_test_VALID_from", CAST('1970-01-01 00:00:00' AS Nullable(DateTime))) AS "test_VALID_from",
+    COALESCE("t_test_VALID_from", CAST('2020-01-01 00:00:00' AS Nullable(DateTime))) AS "test_VALID_from",
     CASE
       WHEN "joined"."_exists" IS NULL
       OR (
diff --git a/tests/core/test_selector.py b/tests/core/test_selector.py
index 3ee4b3d6..f97264fa 100644
--- a/tests/core/test_selector.py
+++ b/tests/core/test_selector.py
@@ -87,7 +87,7 @@ def test_select_models(mocker: MockerFixture, make_snapshot, default_catalog: t.
     local_models[modified_model_v2.fqn] = modified_model_v2.copy(
         update={"mapping_schema": added_model_schema}
     )
-    selector = Selector(state_reader_mock, local_models, default_catalog=default_catalog)
+    selector = Selector(state_reader_mock, local_models, {}, default_catalog=default_catalog)
 
     _assert_models_equal(
         selector.select_models(["db.added_model"], env_name),
@@ -242,7 +242,7 @@ def test_select_models_expired_environment(mocker: MockerFixture, make_snapshot)
 
     local_models: UniqueKeyDict[str, Model] = UniqueKeyDict("models")
     local_models[modified_model_v2.fqn] = modified_model_v2
-    selector = Selector(state_reader_mock, local_models)
+    selector = Selector(state_reader_mock, local_models, {})
 
     _assert_models_equal(
         selector.select_models(["*.modified_model"], env_name, fallback_env_name="prod"),
@@ -298,7 +298,7 @@ def test_select_change_schema(mocker: MockerFixture, make_snapshot):
     local_child = child.copy(update={"mapping_schema": {'"db"': {'"parent"': {"b": "INT"}}}})
     local_models[local_child.fqn] = local_child
 
-    selector = Selector(state_reader_mock, local_models)
+    selector = Selector(state_reader_mock, local_models, {})
 
     selected = selector.select_models(["db.parent"], env_name)
     assert selected[local_child.fqn].data_hash != child.data_hash
@@ -332,7 +332,7 @@ def test_select_models_missing_env(mocker: MockerFixture, make_snapshot):
     local_models: UniqueKeyDict[str, Model] = UniqueKeyDict("models")
     local_models[model.fqn] = model
 
-    selector = Selector(state_reader_mock, local_models)
+    selector = Selector(state_reader_mock, local_models, {})
 
     assert selector.select_models([model.name], "missing_env").keys() == {model.fqn}
     assert not selector.select_models(["missing"], "missing_env")
@@ -508,7 +508,7 @@ def test_expand_model_selections(
         )
         models[model.fqn] = model
 
-    selector = Selector(mocker.Mock(), models)
+    selector = Selector(mocker.Mock(), models, {})
     assert selector.expand_model_selections(selections) == output
 
 
@@ -521,7 +521,7 @@ def test_model_selection_normalized(mocker: MockerFixture, make_snapshot):
         dialect="bigquery",
     )
     models[model.fqn] = model
-    selector = Selector(mocker.Mock(), models, dialect="bigquery")
+    selector = Selector(mocker.Mock(), models, {}, dialect="bigquery")
     assert selector.expand_model_selections(["db.test_Model"]) == {'"db"."test_Model"'}
 
 
@@ -569,7 +569,7 @@ def test_expand_git_selection(
     git_client_mock.list_uncommitted_changed_files.return_value = []
     git_client_mock.list_committed_changed_files.return_value = [model_a._path, model_c._path]
 
-    selector = Selector(mocker.Mock(), models)
+    selector = Selector(mocker.Mock(), models, {})
     selector._git_client = git_client_mock
 
     assert selector.expand_model_selections(expressions) == expected_fqns
@@ -603,7 +603,7 @@ def test_select_models_with_external_parent(mocker: MockerFixture):
     local_models: UniqueKeyDict[str, Model] = UniqueKeyDict("models")
     local_models[added_model.fqn] = added_model
 
-    selector = Selector(state_reader_mock, local_models, default_catalog=default_catalog)
+    selector = Selector(state_reader_mock, local_models, {}, default_catalog=default_catalog)
 
     expanded_selections = selector.expand_model_selections(["+*added_model*"])
     assert expanded_selections == {added_model.fqn}