feat(bigquery): Add enable_change_history option (#1225)

Co-authored-by: Shuting Zhang <[email protected]>

Author: Robin-C

dbt-bigquery/.changes/unreleased/Features-20250811-194447.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Add `enable_change_history` option for models on BigQuery.
+time: 2025-08-11T19:44:47.222876+02:00
+custom:
+    Author: Robin_CHESNE
+    Issue: "1206"

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

@@ -121,6 +121,7 @@ class BigqueryConfig(AdapterConfig):
     intermediate_format: Optional[str] = None
     submission_method: Optional[str] = None
     notebook_template_id: Optional[str] = None
+    enable_change_history: Optional[bool] = None
 
 
 class BigQueryAdapter(BaseAdapter):
@@ -811,6 +812,9 @@ def get_table_options(
             if config.get("partition_expiration_days") is not None:
                 opts["partition_expiration_days"] = config.get("partition_expiration_days")
 
+            if config.get("enable_change_history") is not None:
+                opts["enable_change_history"] = config.get("enable_change_history")
+
             relation_config = getattr(config, "model", None)
             if not temporary and (
                 catalog_relation := self.build_catalog_relation(relation_config)
@@ -827,6 +831,10 @@ def get_table_options(
     @available.parse(lambda *a, **k: {})
     def get_view_options(self, config: Dict[str, Any], node: Dict[str, Any]) -> Dict[str, Any]:
         opts = self.get_common_options(config, node)
+        if config.get("enable_change_history"):
+            raise dbt_common.exceptions.DbtRuntimeError(
+                "`enable_change_history` is not supported for views on BigQuery."
+            )
         return opts
 
     @available.parse(lambda *a, **k: True)

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

@@ -0,0 +1,88 @@
+import pytest
+from dbt.tests.util import run_dbt, run_dbt_and_capture
+import dbt.exceptions
+
+incremental_change_history_enabled_model = """
+{{
+    config(
+        materialized='incremental',
+        enable_change_history=True
+    )
+}}
+select 1 as id
+"""
+
+incremental_change_history_disabled_model = """
+{{
+    config(
+        materialized='incremental',
+        enable_change_history=False
+    )
+}}
+select 1 as id
+"""
+
+table_change_history_enabled_model = """
+{{
+    config(
+        materialized='table',
+        enable_change_history=True
+    )
+}}
+select 1 as id
+"""
+
+view_change_history_enabled_model = """
+{{
+    config(
+        materialized='view',
+        enable_change_history=True
+    )
+}}
+select 1 as id
+"""
+
+
+class TestChangeHistorySuccessful:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "incremental_change_history_enabled.sql": incremental_change_history_enabled_model,
+            "incremental_change_history_disabled.sql": incremental_change_history_disabled_model,
+            "table_change_history_enabled.sql": table_change_history_enabled_model,
+        }
+
+    def get_is_change_history_enabled(self, project, table_name):
+        sql = f"""
+        select is_change_history_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_change_history_options(self, project):
+        results = run_dbt(["run"])
+        assert len(results) == 3
+
+        assert (
+            self.get_is_change_history_enabled(project, "incremental_change_history_enabled")
+            == "YES"
+        )
+        assert (
+            self.get_is_change_history_enabled(project, "incremental_change_history_disabled")
+            == "NO"
+        )
+        assert self.get_is_change_history_enabled(project, "table_change_history_enabled") == "YES"
+
+
+class TestChangeHistoryFail:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "view_with_flag.sql": view_change_history_enabled_model,
+        }
+
+    def test_view_with_flag_fails(self, project):
+        _, stdout = run_dbt_and_capture(["run", "--select", "view_with_flag"], expect_pass=False)
+        assert "`enable_change_history` is not supported for views" in stdout