Add support for non-incremental materialized views

.changes/unreleased/Features-20231108-140752.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Support allow_non_incremental_definition option in BigQuery materialized views
+time: 2023-11-08T14:07:52.28972-05:00
+custom:
+  Author: bnaul,mikealfare
+  Issue: "672"

dbt/adapters/bigquery/connections.py

@@ -322,7 +322,7 @@ def format_rows_number(self, rows_number):
         return f"{rows_number:3.1f}{unit}".strip()
 
     @classmethod
-    def get_google_credentials(cls, profile_credentials) -> GoogleCredentials:
+    def get_google_credentials(cls, profile_credentials) -> GoogleCredentials.Credentials:
         method = profile_credentials.method
         creds = GoogleServiceAccountCredentials.Credentials
 

dbt/adapters/bigquery/impl.py

@@ -115,6 +115,7 @@ class BigqueryConfig(AdapterConfig):
     max_staleness: Optional[str] = None
     enable_list_inference: Optional[bool] = None
     intermediate_format: Optional[str] = None
+    allow_non_incremental_definition: Optional[bool] = None
 
 
 class BigQueryAdapter(BaseAdapter):

dbt/adapters/bigquery/relation.py

@@ -1,19 +1,18 @@
 from dataclasses import dataclass, field
+from itertools import chain, islice
 from typing import FrozenSet, Optional, TypeVar
 
-from itertools import chain, islice
 from dbt.adapters.base.relation import BaseRelation, ComponentName, InformationSchema
+from dbt.adapters.contracts.relation import RelationType, RelationConfig
 from dbt.adapters.relation_configs import RelationConfigChangeAction
+from dbt_common.exceptions import CompilationError
+from dbt_common.utils.dict import filter_null_values
+
 from dbt.adapters.bigquery.relation_configs import (
-    BigQueryClusterConfigChange,
     BigQueryMaterializedViewConfig,
     BigQueryMaterializedViewConfigChangeset,
-    BigQueryOptionsConfigChange,
-    BigQueryPartitionConfigChange,
+    BigQueryRelationConfigChange,
 )
-from dbt.adapters.contracts.relation import RelationType, RelationConfig
-from dbt_common.exceptions import CompilationError
-from dbt_common.utils.dict import filter_null_values
 
 
 Self = TypeVar("Self", bound="BigQueryRelation")
@@ -89,29 +88,67 @@ def materialized_view_config_changeset(
         config_change_collection = BigQueryMaterializedViewConfigChangeset()
         new_materialized_view = cls.materialized_view_from_relation_config(relation_config)
 
-        if new_materialized_view.options != existing_materialized_view.options:
-            config_change_collection.options = BigQueryOptionsConfigChange(
-                action=RelationConfigChangeAction.alter,
-                context=new_materialized_view.options,
+        def add_change(option: str, requires_full_refresh: bool):
+            cls._add_change(
+                config_change_collection=config_change_collection,
+                new_relation=new_materialized_view,
+                existing_relation=existing_materialized_view,
+                option=option,
+                requires_full_refresh=requires_full_refresh,
             )
 
-        if new_materialized_view.partition != existing_materialized_view.partition:
-            # the existing PartitionConfig is not hashable, but since we need to do
-            # a full refresh either way, we don't need to provide a context
-            config_change_collection.partition = BigQueryPartitionConfigChange(
-                action=RelationConfigChangeAction.alter,
-            )
-
-        if new_materialized_view.cluster != existing_materialized_view.cluster:
-            config_change_collection.cluster = BigQueryClusterConfigChange(
-                action=RelationConfigChangeAction.alter,
-                context=new_materialized_view.cluster,
-            )
+        add_change("partition", True)
+        add_change("cluster", True)
+        add_change("enable_refresh", False)
+        add_change("refresh_interval_minutes", False)
+        add_change("max_staleness", False)
+        add_change("allow_non_incremental_definition", True)
+        add_change("kms_key_name", False)
+        add_change("description", False)
+        add_change("labels", False)
 
         if config_change_collection.has_changes:
             return config_change_collection
         return None
 
+    @classmethod
+    def _add_change(
+        cls,
+        config_change_collection,
+        new_relation,
+        existing_relation,
+        option: str,
+        requires_full_refresh: bool,
+    ) -> None:
+        # if there's no change, don't do anything
+        if getattr(new_relation, option) == getattr(existing_relation, option):
+            return
+
+        # determine the type of change: drop, create, alter (includes full refresh)
+        if getattr(new_relation, option) is None:
+            action = RelationConfigChangeAction.drop
+        elif getattr(existing_relation, option) is None:
+            action = RelationConfigChangeAction.create
+        else:
+            action = RelationConfigChangeAction.alter
+
+        # don't worry about passing along the context if it's a going to result in a full refresh
+        if requires_full_refresh:
+            context = None
+        else:
+            context = getattr(new_relation, option)
+
+        # add the change to the collection for downstream processing
+        setattr(
+            config_change_collection,
+            option,
+            BigQueryRelationConfigChange(
+                action=action,
+                context=context,
+                requires_full_refresh=requires_full_refresh,
+            ),
+        )
+
     def information_schema(self, identifier: Optional[str] = None) -> "BigQueryInformationSchema":
         return BigQueryInformationSchema.from_relation(self, identifier)
 

dbt/adapters/bigquery/relation_configs/__init__.py

@@ -1,4 +1,7 @@
-from dbt.adapters.bigquery.relation_configs._base import BigQueryBaseRelationConfig
+from dbt.adapters.bigquery.relation_configs._base import (
+    BigQueryBaseRelationConfig,
+    BigQueryRelationConfigChange,
+)
 from dbt.adapters.bigquery.relation_configs._cluster import (
     BigQueryClusterConfig,
     BigQueryClusterConfigChange,
@@ -7,10 +10,6 @@
     BigQueryMaterializedViewConfig,
     BigQueryMaterializedViewConfigChangeset,
 )
-from dbt.adapters.bigquery.relation_configs._options import (
-    BigQueryOptionsConfig,
-    BigQueryOptionsConfigChange,
-)
 from dbt.adapters.bigquery.relation_configs._partition import (
     PartitionConfig,
     BigQueryPartitionConfigChange,

dbt/adapters/bigquery/relation_configs/_base.py

@@ -1,8 +1,8 @@
 from dataclasses import dataclass
-from typing import Optional, Dict, TYPE_CHECKING
+from typing import Dict, Hashable, Optional, TYPE_CHECKING
 
 from dbt.adapters.base.relation import Policy
-from dbt.adapters.relation_configs import RelationConfigBase
+from dbt.adapters.relation_configs import RelationConfigBase, RelationConfigChangeAction
 from google.cloud.bigquery import Table as BigQueryTable
 from typing_extensions import Self
 
@@ -66,3 +66,10 @@ def _get_first_row(cls, results: "agate.Table") -> "agate.Row":
             import agate
 
             return agate.Row(values=set())
+
+
+@dataclass(frozen=True, eq=True, unsafe_hash=True)
+class BigQueryRelationConfigChange(RelationConfigBase):
+    action: RelationConfigChangeAction
+    context: Optional[Hashable]
+    requires_full_refresh: Optional[bool] = False