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
+  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

@@ -90,9 +90,10 @@ def materialized_view_config_changeset(
         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,
+            # get an options change object with only the options that have changed
+            config_change_collection.options = BigQueryOptionsConfigChange.from_options_configs(
+                new_materialized_view.options,
+                existing_materialized_view.options,
             )
 
         if new_materialized_view.partition != existing_materialized_view.partition:

dbt/adapters/bigquery/relation_configs/_options.py

@@ -1,4 +1,4 @@
-from dataclasses import dataclass
+from dataclasses import asdict, dataclass
 from datetime import datetime, timedelta
 from typing import Any, Dict, Optional
 
@@ -9,6 +9,7 @@
 
 from dbt.adapters.bigquery.relation_configs._base import BigQueryBaseRelationConfig
 from dbt.adapters.bigquery.utility import bool_setting, float_setting, sql_escape
+from dbt.adapters.relation_configs import RelationConfigChangeAction
 
 
 @dataclass(frozen=True, eq=True, unsafe_hash=True)
@@ -22,6 +23,7 @@ class BigQueryOptionsConfig(BigQueryBaseRelationConfig):
     refresh_interval_minutes: Optional[float] = 30
     expiration_timestamp: Optional[datetime] = None
     max_staleness: Optional[str] = None
+    allow_non_incremental_definition: Optional[bool] = None
     kms_key_name: Optional[str] = None
     description: Optional[str] = None
     labels: Optional[Dict[str, str]] = None
@@ -58,6 +60,7 @@ def array(x):
             "refresh_interval_minutes": numeric,
             "expiration_timestamp": interval,
             "max_staleness": interval,
+            "allow_non_incremental_definition": boolean,
             "kms_key_name": string,
             "description": escaped_string,
             "labels": array,
@@ -85,6 +88,7 @@ def from_dict(cls, config_dict: Dict[str, Any]) -> Self:
             "refresh_interval_minutes": float_setting,
             "expiration_timestamp": None,
             "max_staleness": None,
+            "allow_non_incremental_definition": bool_setting,
             "kms_key_name": None,
             "description": None,
             "labels": None,
@@ -101,7 +105,13 @@ def formatted_setting(name: str) -> Any:
         # avoid picking up defaults on dependent options
         # e.g. don't set `refresh_interval_minutes` = 30 when the user has `enable_refresh` = False
         if kwargs_dict["enable_refresh"] is False:
-            kwargs_dict.update({"refresh_interval_minutes": None, "max_staleness": None})
+            kwargs_dict.update(
+                {
+                    "refresh_interval_minutes": None,
+                    "max_staleness": None,
+                    "allow_non_incremental_definition": None,
+                }
+            )
 
         options: Self = super().from_dict(kwargs_dict)  # type: ignore
         return options
@@ -115,6 +125,7 @@ def parse_relation_config(cls, relation_config: RelationConfig) -> Dict[str, Any
                 "refresh_interval_minutes",
                 "expiration_timestamp",
                 "max_staleness",
+                "allow_non_incremental_definition",
                 "kms_key_name",
                 "description",
                 "labels",
@@ -139,7 +150,14 @@ def parse_bq_table(cls, table: BigQueryTable) -> Dict[str, Any]:
             "enable_refresh": table.mview_enable_refresh,
             "refresh_interval_minutes": table.mview_refresh_interval.seconds / 60,
             "expiration_timestamp": table.expires,
-            "max_staleness": None,
+            "max_staleness": (
+                f"INTERVAL '{table._properties.get('maxStaleness')}' YEAR TO SECOND"
+                if table._properties.get("maxStaleness")
+                else None
+            ),
+            "allow_non_incremental_definition": table._properties.get("materializedView", {}).get(
+                "allowNonIncrementalDefinition"
+            ),
             "description": table.description,
         }
 
@@ -158,4 +176,19 @@ class BigQueryOptionsConfigChange(RelationConfigChange):
 
     @property
     def requires_full_refresh(self) -> bool:
-        return False
+        # allow_non_incremental_definition cannot be changed via an ALTER statement
+        return self.context.allow_non_incremental_definition is not None
+
+    @classmethod
+    def from_options_configs(
+        cls, new_options: BigQueryOptionsConfig, existing_options: BigQueryOptionsConfig
+    ) -> Self:
+        new_options_dict = asdict(new_options)
+        existing_options_dict = asdict(existing_options)
+        option_diffs_dict = {
+            k: v
+            for k, v in new_options_dict.items()
+            if new_options_dict[k] != existing_options_dict[k]
+        }
+        option_diffs = BigQueryOptionsConfig.from_dict(option_diffs_dict)
+        return cls(action=RelationConfigChangeAction.alter, context=option_diffs)

tests/functional/adapter/materialized_view_tests/_files.py

@@ -36,7 +36,8 @@
     cluster_by=["id", "value"],
     enable_refresh=True,
     refresh_interval_minutes=60,
-    max_staleness="INTERVAL 45 MINUTE"
+    max_staleness="INTERVAL '0-0 0 0:45:0' YEAR TO SECOND",
+    allow_non_incremental_definition=True
 ) }}
 select
     id,

tests/functional/adapter/materialized_view_tests/_mixin.py

@@ -46,7 +46,7 @@ def my_other_base_table(self, project) -> BaseRelation:
         )
 
     @pytest.fixture(scope="function", autouse=True)
-    def setup(self, project, my_base_table, my_other_base_table, my_materialized_view):  # type: ignore
+    def setup(self, project, my_materialized_view):
         run_dbt(["seed"])
         run_dbt(["run", "--full-refresh"])
 

tests/functional/adapter/materialized_view_tests/test_materialized_view_changes.py

@@ -19,6 +19,8 @@ def check_start_state(project, materialized_view):
         assert isinstance(results, BigQueryMaterializedViewConfig)
         assert results.options.enable_refresh is True
         assert results.options.refresh_interval_minutes == 60
+        assert results.options.max_staleness == "INTERVAL '0-0 0 0:45:0' YEAR TO SECOND"
+        assert results.options.allow_non_incremental_definition is True
         assert results.partition.field == "record_valid_date"
         assert results.partition.data_type == "datetime"
         assert results.partition.granularity == "day"
@@ -27,17 +29,17 @@ def check_start_state(project, materialized_view):
     @staticmethod
     def change_config_via_alter(project, materialized_view):
         initial_model = get_model_file(project, materialized_view)
-        new_model = initial_model.replace("enable_refresh=True", "enable_refresh=False")
+        new_model = initial_model.replace(
+            "refresh_interval_minutes=60", "refresh_interval_minutes=45"
+        )
         set_model_file(project, materialized_view, new_model)
 
     @staticmethod
     def check_state_alter_change_is_applied(project, materialized_view):
         with get_connection(project.adapter):
             results = project.adapter.describe_relation(materialized_view)
         assert isinstance(results, BigQueryMaterializedViewConfig)
-        # these change when run manually
-        assert results.options.enable_refresh is False
-        assert results.options.refresh_interval_minutes == 30  # BQ returns it to the default
+        assert results.options.refresh_interval_minutes == 45
 
     @staticmethod
     def change_config_via_replace(project, materialized_view):
@@ -63,6 +65,9 @@ def change_config_via_replace(project, materialized_view):
             initial_model.replace(old_partition, new_partition)
             .replace("'my_base_table'", "'my_other_base_table'")
             .replace('cluster_by=["id", "value"]', 'cluster_by="id"')
+            .replace(
+                "allow_non_incremental_definition=True", "allow_non_incremental_definition=False"
+            )
         )
         set_model_file(project, materialized_view, new_model)
 
@@ -75,6 +80,7 @@ def check_state_replace_change_is_applied(project, materialized_view):
         assert results.partition.data_type == "int64"
         assert results.partition.range == {"start": 0, "end": 500, "interval": 50}
         assert results.cluster.fields == frozenset({"id"})
+        assert results.options.allow_non_incremental_definition is False
 
 
 class TestBigQueryMaterializedViewChangesApply(