Feature/enable fine grained mutations

dbt-bigquery/src/dbt/adapters/bigquery/impl.py

@@ -122,6 +122,7 @@ class BigqueryConfig(AdapterConfig):
     submission_method: Optional[str] = None
     notebook_template_id: Optional[str] = None
     enable_change_history: Optional[bool] = None
+    enable_fine_grained_mutations: Optional[bool] = None
 
 
 class BigQueryAdapter(BaseAdapter):
@@ -814,6 +815,8 @@ def get_table_options(
 
             if config.get("enable_change_history") is not None:
                 opts["enable_change_history"] = config.get("enable_change_history")
+            if config.get("enable_fine_grained_mutations") is not None:
+                opts["enable_fine_grained_mutations"] = config.get("enable_fine_grained_mutations")
 
             relation_config = getattr(config, "model", None)
             if not temporary and (
@@ -835,6 +838,10 @@ def get_view_options(self, config: Dict[str, Any], node: Dict[str, Any]) -> Dict
             raise dbt_common.exceptions.DbtRuntimeError(
                 "`enable_change_history` is not supported for views on BigQuery."
             )
+        if config.get("enable_fine_grained_mutations"):
+            raise dbt_common.exceptions.DbtRuntimeError(
+                "`enable_fine_grained_mutations` is not supported for views on BigQuery."
+            )
         return opts
 
     @available.parse(lambda *a, **k: True)

dbt-bigquery/tests/functional/adapter/test_fine_grained_mutations.py

@@ -0,0 +1,87 @@
+from typing import Any, Sequence, cast
+
+import pytest  # type: ignore
+from dbt.tests.util import run_dbt, run_dbt_and_capture
+
+
+incremental_fgm_enabled_model = """
+{{
+    config(
+        materialized='incremental',
+        enable_fine_grained_mutations=True
+    )
+}}
+select 1 as id
+"""
+
+
+incremental_fgm_disabled_model = """
+{{
+    config(
+        materialized='incremental',
+        enable_fine_grained_mutations=False
+    )
+}}
+select 1 as id
+"""
+
+
+table_fgm_enabled_model = """
+{{
+    config(
+        materialized='table',
+        enable_fine_grained_mutations=True
+    )
+}}
+select 1 as id
+"""
+
+
+view_fgm_enabled_model = """
+{{
+    config(
+        materialized='view',
+        enable_fine_grained_mutations=True
+    )
+}}
+select 1 as id
+"""
+
+
+class TestFineGrainedMutationsSuccessful:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "incremental_fgm_enabled.sql": incremental_fgm_enabled_model,
+            "incremental_fgm_disabled.sql": incremental_fgm_disabled_model,
+            "table_fgm_enabled.sql": table_fgm_enabled_model,
+        }
+
+    def get_is_fgm_enabled(self, project, table_name):
+        sql = f"""
+        select is_fine_grained_mutations_enabled
+        from `{project.database}.{project.test_schema}.INFORMATION_SCHEMA.TABLES`
+        where table_name = '{table_name}'
+        """
+        results = project.run_sql(sql, fetch="one")
+        return results[0] if results else None
+
+    def test_fgm_options(self, project):
+        results = cast(Sequence[Any], run_dbt(["run"]))
+        assert len(results) == 3
+
+        assert self.get_is_fgm_enabled(project, "incremental_fgm_enabled") == "YES"
+        assert self.get_is_fgm_enabled(project, "incremental_fgm_disabled") == "NO"
+        assert self.get_is_fgm_enabled(project, "table_fgm_enabled") == "YES"
+
+
+class TestFineGrainedMutationsFail:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "view_with_flag.sql": view_fgm_enabled_model,
+        }
+
+    def test_view_with_flag_fails(self):
+        _, stdout = run_dbt_and_capture(["run", "--select", "view_with_flag"], expect_pass=False)
+        assert "`enable_fine_grained_mutations` is not supported for views on BigQuery." in stdout