diff --git a/.changes/unreleased/Features-20231108-140752.yaml b/.changes/unreleased/Features-20231108-140752.yaml new file mode 100644 index 000000000..56dcc6a60 --- /dev/null +++ b/.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" diff --git a/dbt/adapters/bigquery/connections.py b/dbt/adapters/bigquery/connections.py index 14f958a05..ee9de0605 100644 --- a/dbt/adapters/bigquery/connections.py +++ b/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 diff --git a/dbt/adapters/bigquery/impl.py b/dbt/adapters/bigquery/impl.py index dc5cf6e17..7631c40a6 100644 --- a/dbt/adapters/bigquery/impl.py +++ b/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): diff --git a/dbt/adapters/bigquery/relation.py b/dbt/adapters/bigquery/relation.py index 086b4a2aa..19dab336e 100644 --- a/dbt/adapters/bigquery/relation.py +++ b/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) diff --git a/dbt/adapters/bigquery/relation_configs/__init__.py b/dbt/adapters/bigquery/relation_configs/__init__.py index 9ccdec1e0..8b76bbd32 100644 --- a/dbt/adapters/bigquery/relation_configs/__init__.py +++ b/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, diff --git a/dbt/adapters/bigquery/relation_configs/_base.py b/dbt/adapters/bigquery/relation_configs/_base.py index 45e29b99f..67e332d30 100644 --- a/dbt/adapters/bigquery/relation_configs/_base.py +++ b/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 diff --git a/dbt/adapters/bigquery/relation_configs/_materialized_view.py b/dbt/adapters/bigquery/relation_configs/_materialized_view.py index 81ca6b3de..442d31196 100644 --- a/dbt/adapters/bigquery/relation_configs/_materialized_view.py +++ b/dbt/adapters/bigquery/relation_configs/_materialized_view.py @@ -1,32 +1,34 @@ -from dataclasses import dataclass -from typing import Any, Dict, Optional +from dataclasses import dataclass, fields +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, Optional, Set from dbt.adapters.contracts.relation import ( - RelationConfig, ComponentName, + RelationConfig, +) +from dbt.adapters.relation_configs import ( + RelationConfigValidationMixin, + RelationConfigValidationRule, ) +from dbt_common.exceptions import DbtRuntimeError from google.cloud.bigquery import Table as BigQueryTable +from typing_extensions import Self -from dbt.adapters.bigquery.relation_configs._base import BigQueryBaseRelationConfig -from dbt.adapters.bigquery.relation_configs._options import ( - BigQueryOptionsConfig, - BigQueryOptionsConfigChange, -) -from dbt.adapters.bigquery.relation_configs._partition import ( - BigQueryPartitionConfigChange, - PartitionConfig, -) -from dbt.adapters.bigquery.relation_configs._cluster import ( - BigQueryClusterConfig, - BigQueryClusterConfigChange, +from dbt.adapters.bigquery.relation_configs._base import ( + BigQueryBaseRelationConfig, + BigQueryRelationConfigChange, ) +from dbt.adapters.bigquery.relation_configs._cluster import BigQueryClusterConfig +from dbt.adapters.bigquery.relation_configs._partition import PartitionConfig +from dbt.adapters.bigquery.utility import bool_setting, float_setting @dataclass(frozen=True, eq=True, unsafe_hash=True) -class BigQueryMaterializedViewConfig(BigQueryBaseRelationConfig): +class BigQueryMaterializedViewConfig(BigQueryBaseRelationConfig, RelationConfigValidationMixin): """ This config follow the specs found here: https://cloud.google.com/bigquery/docs/reference/standard-sql/data-definition-language#create_materialized_view_statement + https://cloud.google.com/bigquery/docs/reference/standard-sql/data-definition-language#materialized_view_option_list The following parameters are configurable by dbt: - table_id: name of the materialized view @@ -35,64 +37,159 @@ class BigQueryMaterializedViewConfig(BigQueryBaseRelationConfig): - options: options that get set in `SET OPTIONS()` clause - partition: object containing partition information - cluster: object containing cluster information + + The following options are configurable by dbt: + - enable_refresh: turns on/off automatic refresh + - refresh_interval_minutes: the refresh interval in minutes, when enable_refresh is True + - expiration_timestamp: the expiration of data in the underlying table + - max_staleness: the oldest data can be before triggering a refresh + - allow_non_incremental_definition: allows non-incremental reloads, requires max_staleness + - kms_key_name: the name of the keyfile + - description: the comment to add to the materialized view + - labels: labels to add to the materialized view + + Note: + BigQuery allows options to be "unset" in the sense that they do not contain a value (think `None` or `null`). + This can be counterintuitive when that option is a boolean; it introduces a third value, in particular + a value that behaves "false-y". The practice is to mimic the data platform's inputs to the extent + possible to minimize any translation confusion between dbt docs and the platform's (BQ's) docs. + The values `False` and `None` will behave differently when producing the DDL options: + - `False` will show up in the statement submitted to BQ with the value `False` + - `None` will not show up in the statement submitted to BQ at all """ table_id: str dataset_id: str project_id: str - options: BigQueryOptionsConfig partition: Optional[PartitionConfig] = None cluster: Optional[BigQueryClusterConfig] = None + enable_refresh: Optional[bool] = True + 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 + + @property + def validation_rules(self) -> Set[RelationConfigValidationRule]: + # validation_check is what is allowed + return { + RelationConfigValidationRule( + validation_check=self.allow_non_incremental_definition is not True + or self.max_staleness is not None, + validation_error=DbtRuntimeError( + "Please provide a setting for max_staleness when enabling allow_non_incremental_definition.\n" + "Received:\n" + f" allow_non_incremental_definition: {self.allow_non_incremental_definition}\n" + f" max_staleness: {self.max_staleness}\n" + ), + ), + RelationConfigValidationRule( + validation_check=self.enable_refresh is True + or all( + [self.max_staleness is None, self.allow_non_incremental_definition is None] + ), + validation_error=DbtRuntimeError( + "Do not provide a setting for refresh_interval_minutes, max_staleness, nor allow_non_incremental_definition when disabling enable_refresh.\n" + "Received:\n" + f" enable_refresh: {self.enable_refresh}\n" + f" max_staleness: {self.max_staleness}\n" + f" allow_non_incremental_definition: {self.allow_non_incremental_definition}\n" + ), + ), + } @classmethod - def from_dict(cls, config_dict: Dict[str, Any]) -> "BigQueryMaterializedViewConfig": + def from_dict(cls, config_dict: Dict[str, Any]) -> Self: # required kwargs_dict: Dict[str, Any] = { "table_id": cls._render_part(ComponentName.Identifier, config_dict["table_id"]), "dataset_id": cls._render_part(ComponentName.Schema, config_dict["dataset_id"]), "project_id": cls._render_part(ComponentName.Database, config_dict["project_id"]), - "options": BigQueryOptionsConfig.from_dict(config_dict["options"]), } # optional - if partition := config_dict.get("partition"): - kwargs_dict.update({"partition": PartitionConfig.parse(partition)}) + optional_settings = { + "partition": PartitionConfig.parse, + "cluster": BigQueryClusterConfig.from_dict, + "enable_refresh": bool_setting, + "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, + } - if cluster := config_dict.get("cluster"): - kwargs_dict.update({"cluster": BigQueryClusterConfig.from_dict(cluster)}) + for setting, parser in optional_settings.items(): + value = config_dict.get(setting) + if value is not None and parser is not None: + kwargs_dict.update({setting: parser(value)}) # type: ignore + elif value is not None: + kwargs_dict.update({setting: value}) - materialized_view: "BigQueryMaterializedViewConfig" = super().from_dict(kwargs_dict) # type: ignore + materialized_view: Self = super().from_dict(kwargs_dict) # type: ignore return materialized_view @classmethod def parse_relation_config(cls, relation_config: RelationConfig) -> Dict[str, Any]: - config_dict = { + config_extras = relation_config.config.extra # type: ignore + + config_dict: Dict[str, Any] = { + # required "table_id": relation_config.identifier, "dataset_id": relation_config.schema, "project_id": relation_config.database, - # despite this being a foreign object, there will always be options because of defaults - "options": BigQueryOptionsConfig.parse_relation_config(relation_config), + # optional - no transformations + "enable_refresh": config_extras.get("enable_refresh"), + "refresh_interval_minutes": config_extras.get("refresh_interval_minutes"), + "max_staleness": config_extras.get("max_staleness"), + "allow_non_incremental_definition": config_extras.get( + "allow_non_incremental_definition" + ), + "kms_key_name": config_extras.get("kms_key_name"), + "description": config_extras.get("description"), + "labels": config_extras.get("labels"), } - # optional - if relation_config.config and "partition_by" in relation_config.config: - config_dict.update({"partition": PartitionConfig.parse_model_node(relation_config)}) + # optional - transformations + if relation_config.config.get("partition_by"): # type: ignore + config_dict["partition"] = PartitionConfig.parse_model_node(relation_config) + + if relation_config.config.get("cluster_by"): # type: ignore + config_dict["cluster"] = BigQueryClusterConfig.parse_relation_config(relation_config) - if relation_config.config and "cluster_by" in relation_config.config: - config_dict.update( - {"cluster": BigQueryClusterConfig.parse_relation_config(relation_config)} + if hours_to_expiration := config_extras.get("hours_to_expiration"): + config_dict["expiration_timestamp"] = datetime.now(tz=timezone.utc) + timedelta( + hours=hours_to_expiration ) + if relation_config.config.persist_docs and config_extras.get("description"): # type: ignore + config_dict["description"] = config_extras.get("description") + return config_dict @classmethod def parse_bq_table(cls, table: BigQueryTable) -> Dict[str, Any]: - config_dict = { + config_dict: Dict[str, Any] = { + # required "table_id": table.table_id, "dataset_id": table.dataset_id, "project_id": table.project, - # despite this being a foreign object, there will always be options because of defaults - "options": BigQueryOptionsConfig.parse_bq_table(table), + # optional - no transformation + "enable_refresh": table.mview_enable_refresh, + "expiration_timestamp": table.expires, + "allow_non_incremental_definition": table._properties.get("materializedView", {}).get( + "allowNonIncrementalDefinition" + ), + "kms_key_name": getattr( + getattr(table, "encryption_configuration"), "kms_key_name", None + ), + "description": table.description, + "labels": table.labels if table.labels != {} else None, } # optional @@ -102,31 +199,46 @@ def parse_bq_table(cls, table: BigQueryTable) -> Dict[str, Any]: if table.clustering_fields: config_dict.update({"cluster": BigQueryClusterConfig.parse_bq_table(table)}) + if refresh_interval_seconds := table.mview_refresh_interval.seconds: + config_dict.update({"refresh_interval_minutes": refresh_interval_seconds / 60}) + + if max_staleness := table._properties.get("maxStaleness"): + config_dict.update({"max_staleness": f"INTERVAL '{max_staleness}' YEAR TO SECOND"}) + return config_dict @dataclass class BigQueryMaterializedViewConfigChangeset: - options: Optional[BigQueryOptionsConfigChange] = None - partition: Optional[BigQueryPartitionConfigChange] = None - cluster: Optional[BigQueryClusterConfigChange] = None + """ + A collection of changes on a materialized view. + + Note: We don't watch for `expiration_timestamp` because it only gets set on the initial creation. + It would naturally change every time since it's set via `hours_to_expiration`, which would push out + the calculated `expiration_timestamp`. + """ + + partition: Optional[BigQueryRelationConfigChange] = None + cluster: Optional[BigQueryRelationConfigChange] = None + enable_refresh: Optional[BigQueryRelationConfigChange] = None + refresh_interval_minutes: Optional[BigQueryRelationConfigChange] = None + max_staleness: Optional[BigQueryRelationConfigChange] = None + allow_non_incremental_definition: Optional[BigQueryRelationConfigChange] = None + kms_key_name: Optional[BigQueryRelationConfigChange] = None + description: Optional[BigQueryRelationConfigChange] = None + labels: Optional[BigQueryRelationConfigChange] = None @property def requires_full_refresh(self) -> bool: return any( - { - self.options.requires_full_refresh if self.options else False, - self.partition.requires_full_refresh if self.partition else False, - self.cluster.requires_full_refresh if self.cluster else False, - } + [ + getattr(self, field.name).requires_full_refresh + if getattr(self, field.name) + else False + for field in fields(self) + ] ) @property def has_changes(self) -> bool: - return any( - { - self.options if self.options else False, - self.partition if self.partition else False, - self.cluster if self.cluster else False, - } - ) + return any([getattr(self, field.name) is not None for field in fields(self)]) diff --git a/dbt/adapters/bigquery/relation_configs/_options.py b/dbt/adapters/bigquery/relation_configs/_options.py deleted file mode 100644 index f0272df08..000000000 --- a/dbt/adapters/bigquery/relation_configs/_options.py +++ /dev/null @@ -1,161 +0,0 @@ -from dataclasses import dataclass -from datetime import datetime, timedelta -from typing import Any, Dict, Optional - -from dbt.adapters.relation_configs import RelationConfigChange -from dbt.adapters.contracts.relation import RelationConfig -from google.cloud.bigquery import Table as BigQueryTable -from typing_extensions import Self - -from dbt.adapters.bigquery.relation_configs._base import BigQueryBaseRelationConfig -from dbt.adapters.bigquery.utility import bool_setting, float_setting, sql_escape - - -@dataclass(frozen=True, eq=True, unsafe_hash=True) -class BigQueryOptionsConfig(BigQueryBaseRelationConfig): - """ - This config manages materialized view options. See the following for more information: - - https://cloud.google.com/bigquery/docs/reference/standard-sql/data-definition-language#materialized_view_option_list - """ - - enable_refresh: Optional[bool] = True - refresh_interval_minutes: Optional[float] = 30 - expiration_timestamp: Optional[datetime] = None - max_staleness: Optional[str] = None - kms_key_name: Optional[str] = None - description: Optional[str] = None - labels: Optional[Dict[str, str]] = None - - def as_ddl_dict(self) -> Dict[str, Any]: - """ - Reformat `options_dict` so that it can be passed into the `bigquery_options()` macro. - - Options should be flattened and filtered prior to passing into this method. For example: - - the "auto refresh" set of options should be flattened into the root instead of stuck under "auto_refresh" - - any option that comes in set as `None` will be unset; this happens mostly due to config changes - """ - - def boolean(x): - return x - - def numeric(x): - return x - - def string(x): - return f"'{x}'" - - def escaped_string(x): - return f'"""{sql_escape(x)}"""' - - def interval(x): - return x - - def array(x): - return list(x.items()) - - option_formatters = { - "enable_refresh": boolean, - "refresh_interval_minutes": numeric, - "expiration_timestamp": interval, - "max_staleness": interval, - "kms_key_name": string, - "description": escaped_string, - "labels": array, - } - - def formatted_option(name: str) -> Optional[Any]: - value = getattr(self, name) - if value is not None: - formatter = option_formatters[name] - return formatter(value) - return None - - options = { - option: formatted_option(option) - for option, option_formatter in option_formatters.items() - if formatted_option(option) is not None - } - - return options - - @classmethod - def from_dict(cls, config_dict: Dict[str, Any]) -> Self: - setting_formatters = { - "enable_refresh": bool_setting, - "refresh_interval_minutes": float_setting, - "expiration_timestamp": None, - "max_staleness": None, - "kms_key_name": None, - "description": None, - "labels": None, - } - - def formatted_setting(name: str) -> Any: - value = config_dict.get(name) - if formatter := setting_formatters[name]: - return formatter(value) - return value - - kwargs_dict = {attribute: formatted_setting(attribute) for attribute in setting_formatters} - - # 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}) - - options: Self = super().from_dict(kwargs_dict) # type: ignore - return options - - @classmethod - def parse_relation_config(cls, relation_config: RelationConfig) -> Dict[str, Any]: - config_dict = { - option: relation_config.config.extra.get(option) # type: ignore - for option in [ - "enable_refresh", - "refresh_interval_minutes", - "expiration_timestamp", - "max_staleness", - "kms_key_name", - "description", - "labels", - ] - } - - # update dbt-specific versions of these settings - if hours_to_expiration := relation_config.config.extra.get( # type: ignore - "hours_to_expiration" - ): # type: ignore - config_dict.update( - {"expiration_timestamp": datetime.now() + timedelta(hours=hours_to_expiration)} - ) - if not relation_config.config.persist_docs: # type: ignore - del config_dict["description"] - - return config_dict - - @classmethod - def parse_bq_table(cls, table: BigQueryTable) -> Dict[str, Any]: - config_dict = { - "enable_refresh": table.mview_enable_refresh, - "refresh_interval_minutes": table.mview_refresh_interval.seconds / 60, - "expiration_timestamp": table.expires, - "max_staleness": None, - "description": table.description, - } - - # map the empty dict to None - if labels := table.labels: - config_dict.update({"labels": labels}) - - if encryption_configuration := table.encryption_configuration: - config_dict.update({"kms_key_name": encryption_configuration.kms_key_name}) - return config_dict - - -@dataclass(frozen=True, eq=True, unsafe_hash=True) -class BigQueryOptionsConfigChange(RelationConfigChange): - context: BigQueryOptionsConfig - - @property - def requires_full_refresh(self) -> bool: - return False diff --git a/dbt/adapters/bigquery/relation_configs/_partition.py b/dbt/adapters/bigquery/relation_configs/_partition.py index 555aa3664..3c24e129e 100644 --- a/dbt/adapters/bigquery/relation_configs/_partition.py +++ b/dbt/adapters/bigquery/relation_configs/_partition.py @@ -6,6 +6,7 @@ from dbt.adapters.contracts.relation import RelationConfig from dbt_common.dataclass_schema import dbtClassMixin, ValidationError from google.cloud.bigquery.table import Table as BigQueryTable +from typing_extensions import Self @dataclass @@ -80,7 +81,7 @@ def render_wrapped(self, alias: Optional[str] = None): return self.render(alias) @classmethod - def parse(cls, raw_partition_by) -> Optional["PartitionConfig"]: + def parse(cls, raw_partition_by: Dict[str, Any]) -> Optional[Self]: if raw_partition_by is None: return None try: diff --git a/dbt/include/bigquery/macros/relations/materialized_view/alter.sql b/dbt/include/bigquery/macros/relations/materialized_view/alter.sql index e71f869ae..b259f7983 100644 --- a/dbt/include/bigquery/macros/relations/materialized_view/alter.sql +++ b/dbt/include/bigquery/macros/relations/materialized_view/alter.sql @@ -11,8 +11,65 @@ {{ get_replace_sql(existing_relation, relation, sql) }} {% else %} + {%- set _needs_comma = False -%} + alter materialized view {{ relation }} - set {{ bigquery_options(configuration_changes.options.context.as_ddl_dict()) }} + set options( + {% if configuration_changes.enable_refresh -%} + {%- if _needs_comma -%},{%- endif -%} + {%- if configuration_changes.enable_refresh.action == 'drop' -%} + enable_refresh = NULL + {%- else -%} + enable_refresh = {{ configuration_changes.enable_refresh.context }} + {%- endif -%} + {%- set _needs_comma = True -%} + {% endif -%} + {%- if configuration_changes.refresh_interval_minutes -%} + {%- if _needs_comma -%},{%- endif -%} + {%- if configuration_changes.refresh_interval_minutes.action == 'drop' -%} + refresh_interval_minutes = NULL + {%- else -%} + refresh_interval_minutes = {{ configuration_changes.refresh_interval_minutes.context }} + {%- endif -%} + {%- set _needs_comma = True -%} + {% endif -%} + {%- if configuration_changes.max_staleness -%} + {%- if _needs_comma -%},{%- endif -%} + {%- if configuration_changes.max_staleness.action == 'drop' -%} + max_staleness = NULL + {%- else -%} + max_staleness = {{ configuration_changes.max_staleness.context }} + {%- endif -%} + {%- set _needs_comma = True -%} + {% endif -%} + {%- if configuration_changes.kms_key_name -%} + {%- if _needs_comma -%},{%- endif -%} + {%- if configuration_changes.kms_key_name.action == 'drop' -%} + kms_key_name = NULL + {%- else -%} + kms_key_name = '{{ configuration_changes.kms_key_name.context }}' + {%- endif -%} + {%- set _needs_comma = True -%} + {% endif -%} + {%- if configuration_changes.description -%} + {%- if _needs_comma -%},{%- endif -%} + {%- if configuration_changes.description.action == 'drop' -%} + description = NULL + {%- else -%} + description = ""{{ configuration_changes.description.context|tojson|safe }}"" + {%- endif -%} + {%- set _needs_comma = True -%} + {% endif -%} + {%- if configuration_changes.labels -%} + {%- if _needs_comma -%},{%- endif -%} + {%- if configuration_changes.labels.action == 'drop' -%} + labels = NULL + {%- else -%} + labels = {{ configuration_changes.labels.context.items()|list }} + {%- endif -%} + {%- set _needs_comma = True -%} + {%- endif -%} + ) {%- endif %} diff --git a/dbt/include/bigquery/macros/relations/materialized_view/create.sql b/dbt/include/bigquery/macros/relations/materialized_view/create.sql index d3e8c7685..35ec08b4e 100644 --- a/dbt/include/bigquery/macros/relations/materialized_view/create.sql +++ b/dbt/include/bigquery/macros/relations/materialized_view/create.sql @@ -2,10 +2,57 @@ {%- set materialized_view = adapter.Relation.materialized_view_from_relation_config(config.model) -%} - create materialized view if not exists {{ relation }} - {% if materialized_view.partition %}{{ partition_by(materialized_view.partition) }}{% endif %} - {% if materialized_view.cluster %}{{ cluster_by(materialized_view.cluster.fields) }}{% endif %} - {{ bigquery_options(materialized_view.options.as_ddl_dict()) }} + {%- set _needs_comma = False -%} + + create materialized view {{ relation }} + {% if materialized_view.partition -%} + {{ partition_by(materialized_view.partition) }} + {% endif -%} + {%- if materialized_view.cluster -%} + {{ cluster_by(materialized_view.cluster.fields) }} + {%- endif %} + options( + {% if materialized_view.enable_refresh -%} + {%- if _needs_comma -%},{%- endif -%} + enable_refresh = {{ materialized_view.enable_refresh }} + {%- set _needs_comma = True -%} + {% endif -%} + {%- if materialized_view.refresh_interval_minutes -%} + {%- if _needs_comma -%},{%- endif -%} + refresh_interval_minutes = {{ materialized_view.refresh_interval_minutes }} + {%- set _needs_comma = True -%} + {% endif -%} + {%- if materialized_view.expiration_timestamp -%} + {%- if _needs_comma -%},{%- endif -%} + expiration_timestamp = '{{ materialized_view.expiration_timestamp }}' + {%- set _needs_comma = True -%} + {% endif -%} + {%- if materialized_view.max_staleness -%} + {%- if _needs_comma -%},{%- endif -%} + max_staleness = {{ materialized_view.max_staleness }} + {%- set _needs_comma = True -%} + {% endif -%} + {%- if materialized_view.allow_non_incremental_definition -%} + {%- if _needs_comma -%},{%- endif -%} + allow_non_incremental_definition = {{ materialized_view.allow_non_incremental_definition }} + {%- set _needs_comma = True -%} + {% endif -%} + {%- if materialized_view.kms_key_name -%} + {%- if _needs_comma -%},{%- endif -%} + kms_key_name = '{{ materialized_view.kms_key_name }}' + {%- set _needs_comma = True -%} + {% endif -%} + {%- if materialized_view.description -%} + {%- if _needs_comma -%},{%- endif -%} + description = ""{{ materialized_view.description|tojson|safe }}"" + {%- set _needs_comma = True -%} + {% endif -%} + {%- if materialized_view.labels -%} + {%- if _needs_comma -%},{%- endif -%} + labels = {{ materialized_view.labels.items()|list }} + {%- set _needs_comma = True -%} + {%- endif %} + ) as {{ sql }} {% endmacro %} diff --git a/dbt/include/bigquery/macros/relations/materialized_view/replace.sql b/dbt/include/bigquery/macros/relations/materialized_view/replace.sql index 2e4a0b69f..9357dc955 100644 --- a/dbt/include/bigquery/macros/relations/materialized_view/replace.sql +++ b/dbt/include/bigquery/macros/relations/materialized_view/replace.sql @@ -2,10 +2,57 @@ {%- set materialized_view = adapter.Relation.materialized_view_from_relation_config(config.model) -%} - create or replace materialized view if not exists {{ relation }} - {% if materialized_view.partition %}{{ partition_by(materialized_view.partition) }}{% endif %} - {% if materialized_view.cluster %}{{ cluster_by(materialized_view.cluster.fields) }}{% endif %} - {{ bigquery_options(materialized_view.options.as_ddl_dict()) }} + {%- set _needs_comma = False -%} + + create or replace materialized view {{ relation }} + {% if materialized_view.partition -%} + {{ partition_by(materialized_view.partition) }} + {% endif -%} + {%- if materialized_view.cluster -%} + {{ cluster_by(materialized_view.cluster.fields) }} + {%- endif %} + options( + {% if materialized_view.enable_refresh -%} + {%- if _needs_comma -%},{%- endif -%} + enable_refresh = {{ materialized_view.enable_refresh }} + {%- set _needs_comma = True -%} + {% endif -%} + {%- if materialized_view.refresh_interval_minutes -%} + {%- if _needs_comma -%},{%- endif -%} + refresh_interval_minutes = {{ materialized_view.refresh_interval_minutes }} + {%- set _needs_comma = True -%} + {% endif -%} + {%- if materialized_view.expiration_timestamp -%} + {%- if _needs_comma -%},{%- endif -%} + expiration_timestamp = '{{ materialized_view.expiration_timestamp }}' + {%- set _needs_comma = True -%} + {% endif -%} + {%- if materialized_view.max_staleness -%} + {%- if _needs_comma -%},{%- endif -%} + max_staleness = {{ materialized_view.max_staleness }} + {%- set _needs_comma = True -%} + {% endif -%} + {%- if materialized_view.allow_non_incremental_definition -%} + {%- if _needs_comma -%},{%- endif -%} + allow_non_incremental_definition = {{ materialized_view.allow_non_incremental_definition }} + {%- set _needs_comma = True -%} + {% endif -%} + {%- if materialized_view.kms_key_name -%} + {%- if _needs_comma -%},{%- endif -%} + kms_key_name = '{{ materialized_view.kms_key_name }}' + {%- set _needs_comma = True -%} + {% endif -%} + {%- if materialized_view.description -%} + {%- if _needs_comma -%},{%- endif -%} + description = ""{{ materialized_view.description|tojson|safe }}"" + {%- set _needs_comma = True -%} + {% endif -%} + {%- if materialized_view.labels -%} + {%- if _needs_comma -%},{%- endif -%} + labels = {{ materialized_view.labels.items()|list }} + {%- set _needs_comma = True -%} + {%- endif %} + ) as {{ sql }} {% endmacro %} diff --git a/tests/functional/adapter/materialized_view_tests/_files.py b/tests/functional/adapter/materialized_view_tests/_files.py index 86714036a..5bd43a251 100644 --- a/tests/functional/adapter/materialized_view_tests/_files.py +++ b/tests/functional/adapter/materialized_view_tests/_files.py @@ -1,3 +1,6 @@ +# flake8: noqa +# ignores the special characters in the descripton check + MY_SEED = """ id,value,record_valid_date 1,100,2023-01-01 00:00:00 @@ -25,9 +28,10 @@ # the whitespace to the left on partition matters here +# this should test all possible config (skip KMS key since it needs to exist) MY_MATERIALIZED_VIEW = """ {{ config( - materialized='materialized_view', + materialized="materialized_view", partition_by={ "field": "record_valid_date", "data_type": "datetime", @@ -36,7 +40,11 @@ cluster_by=["id", "value"], enable_refresh=True, refresh_interval_minutes=60, - max_staleness="INTERVAL 45 MINUTE" + hours_to_expiration=24, + max_staleness="INTERVAL '0-0 0 0:45:0' YEAR TO SECOND", + allow_non_incremental_definition=True, + description=" 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"]) diff --git a/tests/functional/adapter/materialized_view_tests/test_materialized_view_changes.py b/tests/functional/adapter/materialized_view_tests/test_materialized_view_changes.py index c821c68fc..e61ebf8c9 100644 --- a/tests/functional/adapter/materialized_view_tests/test_materialized_view_changes.py +++ b/tests/functional/adapter/materialized_view_tests/test_materialized_view_changes.py @@ -17,8 +17,10 @@ def check_start_state(project, materialized_view): with get_connection(project.adapter): results = project.adapter.describe_relation(materialized_view) assert isinstance(results, BigQueryMaterializedViewConfig) - assert results.options.enable_refresh is True - assert results.options.refresh_interval_minutes == 60 + assert results.enable_refresh is True + assert results.refresh_interval_minutes == 60 + assert results.max_staleness == "INTERVAL '0-0 0 0:45:0' YEAR TO SECOND" + assert results.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,7 +29,9 @@ 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 @@ -35,9 +39,7 @@ 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.allow_non_incremental_definition is False class TestBigQueryMaterializedViewChangesApply( diff --git a/tests/functional/adapter/materialized_view_tests/test_materialized_view_cluster_changes.py b/tests/functional/adapter/materialized_view_tests/test_materialized_view_cluster_changes.py index 74e174d4f..e7ef5d903 100644 --- a/tests/functional/adapter/materialized_view_tests/test_materialized_view_cluster_changes.py +++ b/tests/functional/adapter/materialized_view_tests/test_materialized_view_cluster_changes.py @@ -19,8 +19,8 @@ def check_start_state(project, materialized_view): with get_connection(project.adapter): results = project.adapter.describe_relation(materialized_view) assert isinstance(results, BigQueryMaterializedViewConfig) - assert results.options.enable_refresh is True - assert results.options.refresh_interval_minutes == 60 + assert results.enable_refresh is True + assert results.refresh_interval_minutes == 60 assert results.cluster.fields == frozenset({"id", "value"}) @staticmethod @@ -34,8 +34,8 @@ 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) - assert results.options.enable_refresh is False - assert results.options.refresh_interval_minutes == 30 # BQ returns it to the default + assert results.enable_refresh is False + assert results.refresh_interval_minutes == 30 # BQ returns it to the default @staticmethod def change_config_via_replace(project, materialized_view): diff --git a/tests/functional/adapter/materialized_view_tests/test_materialized_view_partition_changes.py b/tests/functional/adapter/materialized_view_tests/test_materialized_view_partition_changes.py index 7f396ae1b..deaf3de45 100644 --- a/tests/functional/adapter/materialized_view_tests/test_materialized_view_partition_changes.py +++ b/tests/functional/adapter/materialized_view_tests/test_materialized_view_partition_changes.py @@ -19,8 +19,8 @@ def check_start_state(project, materialized_view): with get_connection(project.adapter): results = project.adapter.describe_relation(materialized_view) assert isinstance(results, BigQueryMaterializedViewConfig) - assert results.options.enable_refresh is True - assert results.options.refresh_interval_minutes == 60 + assert results.enable_refresh is True + assert results.refresh_interval_minutes == 60 assert results.partition.field == "record_valid_date" assert results.partition.data_type == "datetime" assert results.partition.granularity == "day" @@ -37,8 +37,8 @@ def check_state_alter_change_is_applied(project, materialized_view): 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.enable_refresh is False + assert results.refresh_interval_minutes == 30 # BQ returns it to the default @staticmethod def change_config_via_replace(project, materialized_view): diff --git a/tests/functional/adapter/materialized_view_tests/test_materialized_view_refresh_changes.py b/tests/functional/adapter/materialized_view_tests/test_materialized_view_refresh_changes.py new file mode 100644 index 000000000..dbdd792ed --- /dev/null +++ b/tests/functional/adapter/materialized_view_tests/test_materialized_view_refresh_changes.py @@ -0,0 +1,206 @@ +from dbt.adapters.base.relation import BaseRelation +from dbt.adapters.contracts.relation import RelationType +from dbt.tests.util import ( + get_connection, + get_model_file, + run_dbt, + set_model_file, +) +import pytest + +from tests.functional.adapter.materialized_view_tests import _files + + +class RefreshChanges: + @pytest.fixture(scope="class", autouse=True) + def seeds(self): + return {"my_seed.csv": _files.MY_SEED} + + @pytest.fixture(scope="class", autouse=True) + def models(self): + yield { + "my_base_table.sql": _files.MY_BASE_TABLE, + "my_materialized_view.sql": _files.MY_MATERIALIZED_VIEW_INCREMENTAL, + } + + @pytest.fixture(scope="class") + def my_base_table(self, project) -> BaseRelation: + return project.adapter.Relation.create( + identifier="my_base_table", + schema=project.test_schema, + database=project.database, + type=RelationType.Table, + ) + + @pytest.fixture(scope="class") + def my_materialized_view(self, project) -> BaseRelation: + return project.adapter.Relation.create( + identifier="my_materialized_view", + schema=project.test_schema, + database=project.database, + type=RelationType.MaterializedView, + ) + + @staticmethod + def check_start_state(project, my_materialized_view): + raise NotImplementedError + + @staticmethod + def change_config(project, my_materialized_view): + raise NotImplementedError + + @staticmethod + def check_end_state(project, my_materialized_view): + raise NotImplementedError + + def test_refresh_changes(self, project, my_materialized_view): + # arrange + run_dbt(["seed"]) + run_dbt(["run"]) + self.check_start_state(project, my_materialized_view) + + # act + self.change_config(project, my_materialized_view) + run_dbt(["run", "--models", my_materialized_view.identifier]) + + # assert + self.check_end_state(project, my_materialized_view) + + +class TestTurnOnAllowNonIncremental(RefreshChanges): + @pytest.fixture(scope="class", autouse=True) + def models(self): + yield { + "my_base_table.sql": _files.MY_BASE_TABLE, + "my_materialized_view.sql": _files.MY_MINIMAL_MATERIALIZED_VIEW, + } + + @staticmethod + def check_start_state(project, my_materialized_view): + with get_connection(project.adapter): + results = project.adapter.describe_relation(my_materialized_view) + assert results.enable_refresh is True + assert results.refresh_interval_minutes == 30 + assert results.max_staleness is None + assert results.allow_non_incremental_definition is None + + @staticmethod + def change_config(project, my_materialized_view): + initial_model = get_model_file(project, my_materialized_view) + new_model = initial_model.replace( + "materialized='materialized_view'", + """materialized='materialized_view', allow_non_incremental_definition=True, max_staleness="INTERVAL '0-0 0 0:45:0' YEAR TO SECOND" """, + ) + set_model_file(project, my_materialized_view, new_model) + + @staticmethod + def check_end_state(project, my_materialized_view): + with get_connection(project.adapter): + results = project.adapter.describe_relation(my_materialized_view) + assert results.enable_refresh is True + assert results.refresh_interval_minutes == 30 + assert results.max_staleness == "INTERVAL '0-0 0 0:45:0' YEAR TO SECOND" + assert results.allow_non_incremental_definition is True + + +class TestTurnOffAllowNonIncremental(RefreshChanges): + @pytest.fixture(scope="class", autouse=True) + def models(self): + yield { + "my_base_table.sql": _files.MY_BASE_TABLE, + "my_materialized_view.sql": _files.MY_MATERIALIZED_VIEW_NON_INCREMENTAL, + } + + @staticmethod + def check_start_state(project, my_materialized_view): + with get_connection(project.adapter): + results = project.adapter.describe_relation(my_materialized_view) + assert results.enable_refresh is True + assert results.refresh_interval_minutes == 30 + assert results.max_staleness == "INTERVAL '0-0 0 0:45:0' YEAR TO SECOND" + assert results.allow_non_incremental_definition is True + + @staticmethod + def change_config(project, my_materialized_view): + initial_model = get_model_file(project, my_materialized_view) + new_model = initial_model.replace( + "allow_non_incremental_definition=True", "allow_non_incremental_definition=False" + ) + set_model_file(project, my_materialized_view, new_model) + + @staticmethod + def check_end_state(project, my_materialized_view): + with get_connection(project.adapter): + results = project.adapter.describe_relation(my_materialized_view) + assert results.enable_refresh is True + assert results.refresh_interval_minutes == 30 + assert results.max_staleness == "INTERVAL '0-0 0 0:45:0' YEAR TO SECOND" + assert results.allow_non_incremental_definition is False + + +class TestUnsetAllowNonIncremental(RefreshChanges): + @pytest.fixture(scope="class", autouse=True) + def models(self): + yield { + "my_base_table.sql": _files.MY_BASE_TABLE, + "my_materialized_view.sql": _files.MY_MATERIALIZED_VIEW_NON_INCREMENTAL, + } + + @staticmethod + def check_start_state(project, my_materialized_view): + with get_connection(project.adapter): + results = project.adapter.describe_relation(my_materialized_view) + assert results.enable_refresh is True + assert results.refresh_interval_minutes == 30 + assert results.max_staleness == "INTERVAL '0-0 0 0:45:0' YEAR TO SECOND" + assert results.allow_non_incremental_definition is True + + @staticmethod + def change_config(project, my_materialized_view): + initial_model = get_model_file(project, my_materialized_view) + new_model = initial_model.replace("allow_non_incremental_definition=True,", "") + set_model_file(project, my_materialized_view, new_model) + + @staticmethod + def check_end_state(project, my_materialized_view): + with get_connection(project.adapter): + results = project.adapter.describe_relation(my_materialized_view) + assert results.enable_refresh is True + assert results.refresh_interval_minutes == 30 + assert results.max_staleness == "INTERVAL '0-0 0 0:45:0' YEAR TO SECOND" + assert results.allow_non_incremental_definition is None + + +class TestTurnOffRefresh(RefreshChanges): + @pytest.fixture(scope="class", autouse=True) + def models(self): + yield { + "my_base_table.sql": _files.MY_BASE_TABLE, + "my_materialized_view.sql": _files.MY_MATERIALIZED_VIEW_REFRESH_CONFIG, + } + + @staticmethod + def check_start_state(project, my_materialized_view): + with get_connection(project.adapter): + results = project.adapter.describe_relation(my_materialized_view) + assert results.enable_refresh is True + assert results.refresh_interval_minutes == 60 + assert results.max_staleness == "INTERVAL '0-0 0 0:45:0' YEAR TO SECOND" + assert results.allow_non_incremental_definition is True + + @staticmethod + def change_config(project, my_materialized_view): + initial_model = get_model_file(project, my_materialized_view) + new_model = initial_model.replace("enable_refresh=True", "enable_refresh=False") + set_model_file(project, my_materialized_view, new_model) + + @staticmethod + def check_end_state(project, my_materialized_view): + with get_connection(project.adapter): + results = project.adapter.describe_relation(my_materialized_view) + assert results.enable_refresh is False + assert ( + results.refresh_interval_minutes == 30 + ) # this is a defaulted value in BQ, so it doesn't get cleared + assert results.max_staleness is None + assert results.allow_non_incremental_definition is None