From 09543261048c1c3b35601cda1b8e3234954e9bd2 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Tue, 6 Oct 2026 14:52:02 +0200 Subject: [PATCH 1/3] fix(iac): a replaced BigQuery table gets its row access policies and grants back What: each google_bigquery_row_access_policy and each google_bigquery_table_iam_member on a table this module declares (the shared dataset's table grants, and a planned `iam.bind_bq_table` on the module's own table) carries `lifecycle.replace_triggered_by = [google_bigquery_table..id]`. Why: BigQuery deletes a table's row access policies and IAM policy with the table. When tofu replaced a table (partitioning or a key added to a live table, allowed with --allow-data-loss), the policies' table_id was the same string, so tofu planned them as unchanged and the new table stood ungoverned until the next apply. How the trigger was chosen (measured with tofu 1.12 and terraform-provider-google 6.50 against tests/iac/_fake_bigquery.py): - no trigger (main): the table is replaced, the policies plan no-op, and the live table has none after the apply; - `[google_bigquery_table.]` (the whole resource): fixed, but OpenTofu fires a resource reference on every in-place update too, so a label or a retention change replaced both row access policies, two removals the data-loss gate refuses without --allow-data-loss; - `.creation_time`: correct, but `tofu validate` checks a replace_triggered_by attribute against the schema of the resource carrying it, and a table IAM member has no creation_time; - `.id`: unknown in the plan only when a new table is created, valid on both resource types; replaces with the table, no-op on an in-place update. Prior art: OpenTofu's documented replace_triggered_by, and the terraform_data trigger this emitter already uses for partitioning. Validated: the fake BigQuery now serves row access policies and table IAM and deletes both with their table, as BigQuery does. New real-tofu tests show a replaced table gets both policies and its table grant back in the same apply, and an in-place update (label, retention period) leaves them; they fail without the trigger, and the in-place test fails with the whole-resource form. Emit tests pin the lifecycle, and `tofu validate` accepts a row-filtered and a shared-dataset module. --- fluid_build/iac/providers/gcp.py | 65 ++++++++++++- tests/iac/_fake_bigquery.py | 97 +++++++++++++++++-- tests/iac/test_iac_gcp_governance.py | 70 +++++++++++++- tests/iac/test_iac_gcp_governance_plan.py | 109 +++++++++++++++++++++- 4 files changed, 328 insertions(+), 13 deletions(-) diff --git a/fluid_build/iac/providers/gcp.py b/fluid_build/iac/providers/gcp.py index db09988e..197088c4 100644 --- a/fluid_build/iac/providers/gcp.py +++ b/fluid_build/iac/providers/gcp.py @@ -1029,6 +1029,29 @@ def _emit_policy_tags( return refs +def _replaced_with_table(tbl_name: str) -> Dict[str, Any]: + """The ``lifecycle`` of a resource BigQuery deletes with the table ``tbl_name``. + + Deleting a table deletes its row access policies and its IAM policy (BigQuery's + row-level security introduction says so for the policies), and the table that + replaces it starts with neither. ``tofu`` cannot see that in the same apply: the + policy's ``table_id`` is the same string before and after, so it planned the + policy as unchanged and the new table stood ungoverned until the next apply. + ``replace_triggered_by`` recreates it with the table. + + It names the table's ``id``, not the table: a reference to the resource also + fires on every in-place update, so a new label replaced every row access policy + of the table, removals the data-loss gate refuses without ``--allow-data-loss``. + The ``id`` keeps its value through an update and is unknown in the plan only + when a new table is created, which is exactly a replacement. (``creation_time`` + would say the same, but ``tofu validate`` checks a ``replace_triggered_by`` + attribute against the schema of the resource that carries it, and a table IAM + member has no ``creation_time``.) Measured with ``tofu`` 1.12 and + terraform-provider-google 6.50 against ``tests/iac/_fake_bigquery.py``. + """ + return {"replace_triggered_by": [f"google_bigquery_table.{tbl_name}.id"]} + + def _emit_row_policies( resources: Dict[str, Any], policies: Sequence[_gov.RowPolicy], @@ -1036,7 +1059,10 @@ def _emit_row_policies( tbl_name: str, project: Optional[str], ) -> None: - """A ``google_bigquery_row_access_policy`` per row policy of the table.""" + """A ``google_bigquery_row_access_policy`` per row policy of the table. + + Each is recreated when the table is (:func:`_replaced_with_table`). + """ for policy in policies: row: Dict[str, Any] = { "dataset_id": ds_ref, @@ -1044,6 +1070,7 @@ def _emit_row_policies( "policy_id": policy.policy_id, "filter_predicate": policy.predicate, "grantees": list(policy.grantees), + "lifecycle": _replaced_with_table(tbl_name), } if project: row["project"] = project @@ -1218,6 +1245,8 @@ def _emit_bigquery( "table_id": tofu_ref(f"google_bigquery_table.{tbl_name}.table_id"), "role": role, "member": member, + # BigQuery deletes a table's IAM policy with the table. + "lifecycle": _replaced_with_table(tbl_name), } # Cross-project access needs no new schema fields: declare the consumer @@ -1810,26 +1839,56 @@ def _emit_planned_subscription( ] = body +def _module_table( + resources: Mapping[str, Any], cid: str, dataset: Any, table: Any +) -> Optional[str]: + """The name of this module's ``google_bigquery_table`` for ``dataset.table``, if any. + + Derived the way :func:`_emit_bigquery` names the table and its dataset, and + checked against what it wrote, so a planned grant on another product's table + is never tied to this one. + """ + name = safe_ident(f"{cid}_{table}") + body = (resources.get("google_bigquery_table") or {}).get(name) + if not body or body.get("table_id") != table: + return None + ds_name = safe_ident(f"{cid}_{dataset}") + if body.get("dataset_id") == tofu_ref(f"data.google_bigquery_dataset.{ds_name}.dataset_id"): + return name + owned = (resources.get("google_bigquery_dataset") or {}).get(ds_name) or {} + if ( + body.get("dataset_id") == tofu_ref(f"google_bigquery_dataset.{ds_name}.dataset_id") + and owned.get("dataset_id") == dataset + ): + return name + return None + + def _emit_bq_table_iam(resources: Dict[str, Any], action: Mapping[str, Any], cid: str) -> None: """``iam.bind_bq_table`` → ``google_bigquery_table_iam_member`` (table-scoped IAM). Dataset-level IAM is folded into the dataset ``access`` block by the - ``exposes[]`` walk; this adds the finer table-level grants. + ``exposes[]`` walk; this adds the finer table-level grants. A grant on a + table this module declares is recreated with it (:func:`_replaced_with_table`). """ dataset = action.get("dataset") table = action.get("table") if not (dataset and table): return + tbl_name = _module_table(resources, cid, dataset, table) action_grants = grants_from_legacy_policies(action.get("policies")) for role, grant in role_grants(action_grants, _BQ_TABLE_IAM_ROLES): member = _gcs_member(grant) name = safe_ident(f"{cid}_{dataset}_{table}_{role}_{member}") - resources.setdefault("google_bigquery_table_iam_member", {})[name] = { + body: Dict[str, Any] = { "dataset_id": dataset, "table_id": table, "role": role, "member": member, } + if tbl_name is not None: + body["lifecycle"] = _replaced_with_table(tbl_name) + resources.setdefault("google_bigquery_table_iam_member", {})[name] = body def _emit_composer_dag(resources: Dict[str, Any], action: Mapping[str, Any], cid: str) -> None: diff --git a/tests/iac/_fake_bigquery.py b/tests/iac/_fake_bigquery.py index e6f03f93..23a67e01 100644 --- a/tests/iac/_fake_bigquery.py +++ b/tests/iac/_fake_bigquery.py @@ -18,13 +18,16 @@ terraform-provider-google (the emulator returns a dataset without ``selfLink``, and ``resource_bigquery_dataset.go`` asserts it is a string), so the emulator cannot hold the state a governance test needs. This server stores exactly what -the provider sends for datasets and tables and returns it with the output-only -fields the provider reads (ids, ``selfLink``, ``etag``, timestamps), and a new -dataset gets the default access entries BigQuery gives one (the project's owners, -writers and readers, and the creator). Nothing else is emulated: no query, no -load, no IAM evaluation, no Cloud KMS and no Data Catalog. What it proves is what -the real provider and ``tofu`` plan for a live table, which is where the -data-loss gate decides; what BigQuery itself accepts is not proven here. +the provider sends for datasets, tables, a table's row access policies and its IAM +policy, and returns it with the output-only fields the provider reads (ids, +``selfLink``, ``etag``, timestamps), and a new dataset gets the default access +entries BigQuery gives one (the project's owners, writers and readers, and the +creator). Deleting a table deletes its row access policies and its IAM policy, as +BigQuery does, and a new table under the same id starts with neither. Nothing else is +emulated: no query, no load, no IAM evaluation, no Cloud KMS and no Data Catalog. +What it proves is what the real provider and ``tofu`` plan for a live table, which +is where the data-loss gate decides; what BigQuery itself accepts is not proven +here. The provider is pointed at it with ``bigquery_custom_endpoint`` and a dummy ``access_token``, so no credential and no Google endpoint is involved. @@ -43,6 +46,12 @@ _DATASET_RE = re.compile(r"^/bigquery/v2/projects/([^/]+)/datasets/?([^/]*)$") _TABLE_RE = re.compile(r"^/bigquery/v2/projects/([^/]+)/datasets/([^/]+)/tables/?([^/]*)$") +_ROW_POLICY_RE = re.compile( + r"^/bigquery/v2/projects/([^/]+)/datasets/([^/]+)/tables/([^/]+)/rowAccessPolicies/?([^/]*)$" +) +_TABLE_IAM_RE = re.compile( + r"^/bigquery/v2/projects/([^/]+)/datasets/([^/]+)/tables/([^/:]+):(getIamPolicy|setIamPolicy)$" +) #: What BigQuery puts on a dataset created without ``access`` (BigQuery #: documentation, "Dataset access": the project's basic roles and the creator). @@ -71,11 +80,16 @@ def _legacy_access(dataset: Dict[str, Any]) -> None: class FakeBigQuery: - """Datasets and tables by ``(project, dataset[, table])``; every request is logged.""" + """Datasets, tables and their policies by ``(project, dataset[, table])``; every + request is logged.""" def __init__(self) -> None: self.datasets: Dict[Tuple[str, str], Dict[str, Any]] = {} self.tables: Dict[Tuple[str, str, str], Dict[str, Any]] = {} + #: Row access policies by table, then by policy id. + self.row_policies: Dict[Tuple[str, str, str], Dict[str, Dict[str, Any]]] = {} + #: A table's IAM policy (``bindings``), by table. + self.table_iam: Dict[Tuple[str, str, str], Dict[str, Any]] = {} self.requests: List[Tuple[str, str]] = [] self._lock = threading.Lock() self._server: Optional[ThreadingHTTPServer] = None @@ -162,6 +176,12 @@ def _stamp(self, body: Dict[str, Any]) -> None: body["etag"] = f"etag-{now}" def _route(self, method: str, path: str, body: Dict[str, Any]) -> Tuple[int, Any]: + match = _ROW_POLICY_RE.match(path) + if match: + return self._row_policy(method, *match.groups(), body) + match = _TABLE_IAM_RE.match(path) + if match: + return self._table_iam(method, *match.groups(), body) match = _TABLE_RE.match(path) if match: return self._table(method, *match.groups(), body) @@ -246,10 +266,71 @@ def _table( self.tables[key] = merged return 200, merged if method == "DELETE": + # BigQuery deletes the table's row access policies and IAM policy with it. del self.tables[key] + self.row_policies.pop(key, None) + self.table_iam.pop(key, None) return 204, None return self._not_found(path_of(method, project, dataset, table)) + def _row_policy( + self, + method: str, + project: str, + dataset: str, + table: str, + policy: str, + body: Dict[str, Any], + ) -> Any: + key = (project, dataset, table) + if key not in self.tables: + return self._not_found(f"Table {project}:{dataset}.{table}") + policies = self.row_policies.setdefault(key, {}) + if method == "POST" and not policy: + ref = dict(body.get("rowAccessPolicyReference") or {}) + ref.update(projectId=project, datasetId=dataset, tableId=table) + policy_id = ref.get("policyId") or "" + if policy_id in policies: + return 409, {"error": {"code": 409, "message": "Already Exists"}} + stored = copy.deepcopy(body) + stored["rowAccessPolicyReference"] = ref + self._stamp(stored) + policies[policy_id] = stored + return 200, stored + if method == "GET" and not policy: + return 200, {"rowAccessPolicies": list(policies.values())} + if policy not in policies: + return self._not_found(f"Row access policy {policy} on {project}:{dataset}.{table}") + if method == "GET": + return 200, policies[policy] + if method in ("PUT", "PATCH"): + stored = policies[policy] + merged = dict(stored) if method == "PATCH" else {} + merged.update(copy.deepcopy(body)) + merged["rowAccessPolicyReference"] = stored["rowAccessPolicyReference"] + merged["creationTime"] = stored["creationTime"] + self._stamp(merged) + policies[policy] = merged + return 200, merged + if method == "DELETE": + del policies[policy] + return 204, None + return self._not_found(path_of(method, project, dataset, table, policy)) + + def _table_iam( + self, method: str, project: str, dataset: str, table: str, verb: str, body: Dict[str, Any] + ) -> Any: + key = (project, dataset, table) + if key not in self.tables: + return self._not_found(f"Table {project}:{dataset}.{table}") + if method != "POST": + return self._not_found(path_of(method, project, dataset, f"{table}:{verb}")) + if verb == "setIamPolicy": + stored = copy.deepcopy(body.get("policy") or {}) + stored["etag"] = f"etag-{int(time.time() * 1000)}" + self.table_iam[key] = stored + return 200, self.table_iam.get(key, {"etag": "etag-empty"}) + def path_of(method: str, *parts: str) -> str: return f"{method} {'/'.join(parts)}" diff --git a/tests/iac/test_iac_gcp_governance.py b/tests/iac/test_iac_gcp_governance.py index ffda35d0..b1a9c646 100644 --- a/tests/iac/test_iac_gcp_governance.py +++ b/tests/iac/test_iac_gcp_governance.py @@ -600,6 +600,60 @@ def test_fluid_validate_reports_each_refusal_at_stage_2(): ) == ([], []) +# ── What BigQuery deletes with a table is recreated with it ─────────────── + +CONSENT = {"principal": ANALYSTS, "name": "analysts_consented", "where": "status = 'active'"} +_TRIGGER = {"replace_triggered_by": [f"{TABLE}.id"]} + + +def _row_filtered(**kwargs: Any) -> Dict[str, Any]: + contract = _contract(principals=MAPPING, **kwargs) + contract["exposes"][0].setdefault("policy", {}).setdefault("authz", {})["rowFilters"] = [ + dict(CONSENT) + ] + return contract + + +def _shared_dataset() -> Dict[str, Any]: + contract = _contract(principals=MAPPING) + contract["packaging"] = {"mode": "shared", "pool": "northwind-pool"} + return contract + + +def test_each_row_access_policy_is_recreated_with_its_table(): + """BigQuery deletes a table's row access policies with it; tofu, seeing the same + ``table_id``, kept them in state and the new table stood unfiltered.""" + policies = _resources(_row_filtered())["google_bigquery_row_access_policy"] + assert len(policies) == 2 + for policy in policies.values(): + assert policy["lifecycle"] == _TRIGGER + + +def test_each_table_grant_in_a_shared_dataset_is_recreated_with_its_table(): + members = _resources(_shared_dataset())["google_bigquery_table_iam_member"] + assert members + for member in members.values(): + assert member["lifecycle"] == _TRIGGER + + +def test_a_planned_table_grant_follows_the_table_only_when_this_module_declares_it(): + """``iam.bind_bq_table`` names its table as plain strings; it is tied to the table + resource only when that resource is this module's table of the same dataset.""" + plugin = get_iac_plugin("gcp") + policies = {"readers": {"principals": ["analyst@example.com"], "permissions": ["read"]}} + + def members(dataset: str, table: str) -> List[Dict[str, Any]]: + action = {"op": "iam.bind_bq_table", "dataset": dataset, "table": table} + resources = plugin.emit(_contract(principals=MAPPING), [{**action, "policies": policies}]) + return list(resources["google_bigquery_table_iam_member"].values()) + + (own,) = members("demo_gold", "retention_candidates") + assert own["lifecycle"] == _TRIGGER + for dataset, table in (("demo_gold", "other_table"), ("other_dataset", "retention_candidates")): + (other,) = members(dataset, table) + assert "lifecycle" not in other, (dataset, table) + + # ── tofu validate ───────────────────────────────────────────────────────── _TOFU = shutil.which("tofu") @@ -610,7 +664,14 @@ def test_fluid_validate_reports_each_refusal_at_stage_2(): @pytest.mark.skipif(_TOFU is None, reason="tofu not on PATH") @pytest.mark.parametrize( "shape", - ["all", "existing-key-partition-column", "grants-only", "multi-region-product-key"], + [ + "all", + "existing-key-partition-column", + "grants-only", + "multi-region-product-key", + "row-filters", + "shared-dataset", + ], ) def test_governed_modules_pass_tofu_validate(shape, tmp_path): from tests.iac.test_iac_tofu_validate import _tofu_init_or_skip @@ -633,6 +694,13 @@ def test_governed_modules_pass_tofu_validate(shape, tmp_path): "multi-region-product-key": lambda: _contract( principals=MAPPING, encryption={"kms": "product"}, region="EU" ), + # The row access policies' and the table grants' replace_triggered_by: + # ``tofu validate`` checks its attribute against the carrying resource. + "row-filters": lambda: _row_filtered( + lifecycle={"retention": "P90D", "expire": True}, + restrictions=TestColumnRestrictions.DENY, + ), + "shared-dataset": _shared_dataset, }[shape]() (tmp_path / "main.tf.json").write_text( build_module(get_iac_plugin("gcp"), copy.deepcopy(contract)) diff --git a/tests/iac/test_iac_gcp_governance_plan.py b/tests/iac/test_iac_gcp_governance_plan.py index 5d2ca21e..1dc41a54 100644 --- a/tests/iac/test_iac_gcp_governance_plan.py +++ b/tests/iac/test_iac_gcp_governance_plan.py @@ -23,7 +23,9 @@ The state is made by a real ``tofu apply`` against ``_fake_bigquery.FakeBigQuery``, an in-process stand-in for the BigQuery REST API (the goccy emulator crashes the provider on apply). It also shows the dataset grants are added to the dataset's -access list, not written over it. No Google credential or endpoint is involved. +access list, not written over it, and that a replaced table gets back, in the same +apply, the row access policies and table grants BigQuery deletes with it (the fake +deletes them too). No Google credential or endpoint is involved. Not proven here: what real BigQuery accepts (a load into a policy-tagged column, the service agent's use of the key), Cloud KMS and Data Catalog resources (the @@ -408,3 +410,108 @@ def test_an_unchanged_contract_leaving_the_authoritative_list_revokes_nothing( ) plan = _plan(tmp_path, tofu_env) assert plan["summary"]["remove"] == 0 + + +# ── What BigQuery deletes with a table comes back with it ───────────────── + +ROW_POLICIES = "google_bigquery_row_access_policy" +TABLE_MEMBERS = "google_bigquery_table_iam_member" +ANALYSTS = "group:analysts@company.example" + + +def _row_filtered(**kwargs: Any) -> Dict[str, Any]: + """The plan contract with a row filter: the analysts read the consented rows.""" + contract = _contract(**kwargs) + exposure = contract["exposes"][0] + exposure["binding"]["principals"][ANALYSTS] = "group:analysts@fluid-fake.test-corp.com" + exposure["contract"]["schema"].append({"name": "consent", "type": "boolean"}) + exposure["policy"] = { + "authz": { + "rowFilters": [ + {"principal": ANALYSTS, "name": "analysts_consented", "where": "consent = true"} + ] + } + } + contract["accessPolicy"]["grants"].append({"principal": ANALYSTS, "permissions": ["read"]}) + return contract + + +def _addresses(plan: Dict[str, Any], kind: str) -> List[str]: + return sorted(a for a in plan["changes"] if a.startswith(f"{kind}.")) + + +@pytest.mark.skipif(_SKIP, reason="needs `tofu` on PATH") +def test_a_replaced_table_gets_its_row_access_policies_back(tmp_path, fake, tofu_env): + """BigQuery deletes a table's row access policies with the table. Before + ``replace_triggered_by``, tofu planned them as unchanged (their ``table_id`` + is the same string), and the new table stood with none: every reader the + filter restricts read every row until the next apply.""" + _live(_row_filtered(), tmp_path, fake, tofu_env) + key = (PROJECT, "sales", "orders") + assert sorted(fake.row_policies[key]) == ["analysts_consented", "fluid_all_rows"] + + _write(_row_filtered(retention="P30D"), tmp_path, fake.endpoint) + plan = _plan(tmp_path, tofu_env) + assert _actions(plan, TABLE) == ["delete", "create"] + policies = _addresses(plan, ROW_POLICIES) + assert len(policies) == 2 + for address in policies: + assert _actions(plan, address) == ["delete", "create"], address + # Still gated: the table's replacement loses its rows. + assert _data_loss_blocked(plan["summary"], allow_data_loss=False) + + _apply(tmp_path, tofu_env) + assert sorted(fake.row_policies[key]) == ["analysts_consented", "fluid_all_rows"] + consented = fake.row_policies[key]["analysts_consented"] + assert consented["filterPredicate"] == "consent = true" + assert consented["grantees"] == ["group:analysts@fluid-fake.test-corp.com"] + + +@pytest.mark.skipif(_SKIP, reason="needs `tofu` on PATH") +@pytest.mark.parametrize("change", ["retention-period", "label"]) +def test_an_in_place_table_update_leaves_its_row_access_policies(tmp_path, fake, tofu_env, change): + """The trigger is the table's ``id``, not the table: a reference to the resource + fires on every update too, and a new label then replaced both policies, two + removals the data-loss gate refuses without --allow-data-loss.""" + _live(_row_filtered(retention="P30D"), tmp_path, fake, tofu_env) + after = _row_filtered(retention="P90D" if change == "retention-period" else "P30D") + if change == "label": + after["labels"] = {"team": "sales"} + _write(after, tmp_path, fake.endpoint) + plan = _plan(tmp_path, tofu_env) + assert _actions(plan, TABLE) == ["update"] + for address in _addresses(plan, ROW_POLICIES): + assert _actions(plan, address) == ["no-op"], address + assert plan["summary"]["remove"] == 0 + assert not _data_loss_blocked(plan["summary"], allow_data_loss=False) + + +def _shared(**kwargs: Any) -> Dict[str, Any]: + """The plan contract in a platform-owned dataset: its grants are table IAM members.""" + contract = _contract(**kwargs) + contract["packaging"] = {"mode": "shared", "pool": "sales-pool"} + return contract + + +@pytest.mark.skipif(_SKIP, reason="needs `tofu` on PATH") +def test_a_replaced_table_in_a_shared_dataset_gets_its_grants_back(tmp_path, fake, tofu_env): + """BigQuery deletes a table's IAM policy with the table, so the readers a + shared dataset grants at table level lost their access to the new one.""" + fake._route( + "POST", + f"/bigquery/v2/projects/{PROJECT}/datasets", + {"datasetReference": {"datasetId": "sales"}, "location": "europe-west1"}, + ) + _live(_shared(), tmp_path, fake, tofu_env) + key = (PROJECT, "sales", "orders") + reader = "group:readers@fluid-fake.test-corp.com" + assert any(reader in b.get("members", []) for b in fake.table_iam[key].get("bindings", [])) + + _write(_shared(retention="P30D"), tmp_path, fake.endpoint) + plan = _plan(tmp_path, tofu_env) + assert _actions(plan, TABLE) == ["delete", "create"] + (member,) = _addresses(plan, TABLE_MEMBERS) + assert _actions(plan, member) == ["delete", "create"] + + _apply(tmp_path, tofu_env) + assert any(reader in b.get("members", []) for b in fake.table_iam[key].get("bindings", [])) From 0739491b84f1ee262c69388ba3071027f4cbbce3 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Tue, 6 Oct 2026 14:52:02 +0200 Subject: [PATCH 2/3] fix(gcp): a BigQuery replace or rollback keeps the table's governance What: nothing forge-cli runs recreates a governed BigQuery table to replace its rows any more. - `fluid rollback` (cli/rollback.py::_restore_bigquery) restores with one query job, `SELECT * FROM ` into the live table, write disposition from `_bigquery_load.write_disposition("full_refresh")` (WRITE_TRUNCATE_DATA, WRITE_TRUNCATE on an emulator), CREATE_NEVER, client from `_bigquery_load.bigquery_client`. The destination is a TableReference of the identifiers `_validated_bq_fqn` checked; the recorded ddl[] is still never run and the state file stays untrusted. Before writing, it counts the rows the restoring identity can read against the backup's own numRows and refuses (`rollback_bigquery_backup_filtered`) when a row access policy shows it only some, which would otherwise restore that subset. - `GcpProvider.restore_ddl` returns [] as AWS does. - The GCP planner's replace emits a SQL build as the SELECT, a structured destination and WRITE_TRUNCATE_DATA / CREATE_NEVER; it keeps the table's `bq.ensure_table`; SQL that names its own sink passes through untouched. - The pre-replace backup is `CREATE SNAPSHOT TABLE IF NOT EXISTS ... CLONE ... OPTIONS(expiration_timestamp = now + 30 days)`, not a plain CTAS copy with no row access policy and no policy tag. The "BigQuery has no CLONE" comment is corrected. - BasePlanner, GcpPlanner and provider docstrings; docs/apply.md (what replace and replace-and-build do on the OpenTofu engine), docs/governance-parity.md (dbt's table materialisation still recreates the table: a known gap, follow-up), AGENTS.md, scaffold_ide.py, CHANGELOG. Why: BigQuery documents that CREATE OR REPLACE TABLE drops every row access policy of the table, and the new table takes its schema from the SELECT, so the column policy tags (column-level security and masking) went too. Every filtered or restricted principal then read every row and column until the next apply. The rollback restore is the path that executes on main; the planner's CTAS and backup are latent since the OpenTofu cutover but wrong. Prior art: dlt's BigQuery destination does its atomic replace as one WRITE_TRUNCATE_DATA job into the existing table (dlt-hub/dlt, destinations/impl/bigquery), which forge-cli's load path already follows; BigQuery's REST reference (JobConfigurationQuery.writeDisposition) says WRITE_TRUNCATE_DATA "overwrites the data, but keeps the constraints and schema of the existing table". dbt-bigquery, Dataform and SQLMesh all create-or-replace and re-apply tags afterwards (an exposure window, and no row access policies), so their pattern was not copied. Validated: the new and rewritten tests (tests/providers/ test_gcp_replace_keeps_governance.py, tests/cli/test_rollback.py:: TestRestoreBigQuery, with the IaC tests: 37) go 22 failed / 15 passed with the source fix stashed and 37 passed with it restored (the 15 are invariants that hold on main too). Full suite without tofu on PATH: 20067 passed, 660 skipped, rc=0, with six dbt tests deselected that fail identically on origin/main (dbt-duckdb is not in the venv). ruff, black 24.10.0, the product guardrail and detect-secrets pass. .secrets.baseline only moves one AGENTS.md line number (the entry set is unchanged, 250 before and after). A live probe of a WRITE_TRUNCATE_DATA query job on real BigQuery (tags, policies, column mapping, snapshot governance) is written but not run here. --- .secrets.baseline | 4 +- AGENTS.md | 2 + CHANGELOG.md | 27 ++ docs/apply.md | 36 +- docs/governance-parity.md | 43 ++- fluid_build/cli/_rollback_writer.py | 9 +- fluid_build/cli/rollback.py | 190 +++++++--- fluid_build/cli/scaffold_ide.py | 7 + fluid_build/providers/_planner_base.py | 30 +- fluid_build/providers/gcp/plan/planner.py | 205 ++++++---- fluid_build/providers/gcp/provider.py | 50 +-- tests/cli/test_rollback.py | 208 +++++++--- .../test_gcp_replace_keeps_governance.py | 357 ++++++++++++++++++ 13 files changed, 921 insertions(+), 247 deletions(-) create mode 100644 tests/providers/test_gcp_replace_keeps_governance.py diff --git a/.secrets.baseline b/.secrets.baseline index c4cc8d9a..820e1377 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -157,7 +157,7 @@ "filename": "AGENTS.md", "hashed_secret": "9d4e1e23bd5b727046a9e3b4b7db57bd8d6ee684", "is_verified": false, - "line_number": 597 + "line_number": 599 } ], "Jenkinsfile": [ @@ -2154,5 +2154,5 @@ } ] }, - "generated_at": "2026-10-05T22:13:49Z" + "generated_at": "2026-10-06T12:23:50Z" } diff --git a/AGENTS.md b/AGENTS.md index 2f733ef2..f638bf86 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -51,6 +51,8 @@ Every contract is delivered through an 11-stage pipeline. Each stage is a hard g `--allow-data-loss` is required for `replace*` when `FLUID_ENV != dev` OR target has rows. `--no-verify-plan-binding` is an emergency DR waiver for plan-binding checks (logged at WARNING level so audit trails catch it). +The `replace` rows describe the native engine. On the OpenTofu engine (aws, gcp, snowflake, confluent) `replace` emits the same module as `amend` and adds the data-loss gate; a table is replaced only where `tofu` plans it, and on GCP its row access policies and table grants are recreated with it (`replace_triggered_by`). A BigQuery table is never recreated to replace its rows: the planner's replace and `fluid rollback` run a `WRITE_TRUNCATE_DATA` query job into the existing table (a `CREATE OR REPLACE TABLE` drops its row access policies and policy tags), and the pre-replace backup is a `CREATE SNAPSHOT TABLE … CLONE`. dbt's `table` materialisation still recreates a governed table (known gap; `docs/governance-parity.md`). See `docs/apply.md`. + ### Plan-Binding (Terraform-Style "Apply Consumes Exact Plan") Stage 6 `fluid plan` emits two cryptographic fields in `plan.json`: diff --git a/CHANGELOG.md b/CHANGELOG.md index 7425c398..032b085f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,33 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 now builds the tag's commit, also when a manual run starts from `main`, and the quality gate checks `ci.yml` on that commit. +### Fixed + +- **A BigQuery replace or rollback keeps the table's governance.** Nothing forge-cli + runs recreates a governed BigQuery table any more. `CREATE OR REPLACE TABLE … AS + SELECT`, which BigQuery documents drops every row access policy, and whose SELECT + schema has no policy tag, left every filtered or restricted principal reading every + row and column until the next apply. dbt's `table` materialisation still does (a + known gap, documented in `docs/governance-parity.md`). + - `fluid rollback` restores with one query job, `SELECT * FROM ` into the + live table with `WRITE_TRUNCATE_DATA` and `CREATE_NEVER` (`WRITE_TRUNCATE` on an + emulator). The destination is a table reference built from the validated + identifiers, and the recorded `ddl[]` is still never run. A backup the restoring + identity cannot read in full, because a row access policy shows it only some rows, + is refused (`rollback_bigquery_backup_filtered`) instead of restoring that subset. + - `GcpProvider.restore_ddl` records no restore SQL, as AWS does. + - The GCP planner's replace emits a SQL build as the SELECT with a structured + destination and `WRITE_TRUNCATE_DATA` / `CREATE_NEVER`, keeps the table's + `bq.ensure_table`, and backs the table up with a zero-copy + `CREATE SNAPSHOT TABLE … CLONE` that expires after 30 days, not a plain copy with + no policy and no tag. SQL that names its own sink passes through as before. + - When `tofu` replaces a BigQuery table (partitioning or a key added to a live + table), its row access policies and table IAM members are recreated with it in + the same apply (`lifecycle.replace_triggered_by` on the table's `id`). BigQuery + deletes them with the table, and `tofu` planned them as unchanged. An in-place + update of the table leaves them alone, so a label change still needs no + `--allow-data-loss`. + ## [0.19.0] — 2026-10-06 A contract can now say that a principal reads a column masked rather than not at all, diff --git a/docs/apply.md b/docs/apply.md index dae8bdc9..2b122ba6 100644 --- a/docs/apply.md +++ b/docs/apply.md @@ -21,7 +21,7 @@ fluid apply contract.fluid.yaml --mode amend --yes | `create-only` | `CREATE IF NOT EXISTS`, fails if the target exists | no | untouched | no | | `amend` *(default)* | additive — `ALTER … ADD COLUMN IF NOT EXISTS`; views `CREATE OR REPLACE` | no | preserved; new columns `NULL` | no | | `amend-and-build` | same as `amend` | yes | preserved; transforms re-run | no | -| `replace` | drop + recreate the target | no | **dropped** | **yes** | +| `replace` | drop + recreate the target (native engine; see [the OpenTofu engine](#apply-engines-and-what-that-changes)) | no | **dropped** | **yes** | | `replace-and-build` | same as `replace` | yes (`dbt --full-refresh`) | **dropped**, then rebuilt | **yes** | `--mode` defaults to `amend`. `--dry-run` is an ergonomic alias for @@ -64,8 +64,8 @@ The engine is resolved per provider — there is no user-facing switch. | `aws`, `gcp`, `snowflake`, `confluent` | OpenTofu (`.tf.json` + `tofu init/plan/apply`) | | `local` | native in-process apply | -Two mode behaviours differ by engine. Both are reported at run time, but -know them before you plan a destructive change: +Three mode behaviours differ by engine. Know them before you plan a +destructive change: **1. No pre-replace snapshot on the OpenTofu engine.** The native path plans a pre-flight zero-copy snapshot (`__backup_`) and records @@ -91,6 +91,36 @@ matches the live table will therefore plan clean — including under `fluid verify --strict` is the gate — it exits non-zero on a type mismatch. +**3. `replace` provisions what `amend` provisions.** The OpenTofu engine +emits the same module in every mode, so `--mode replace` drops and +recreates nothing by itself: it adds the data-loss gate and the drift +report to a normal `tofu apply`. A table is replaced only when `tofu` +plans it (a change the platform cannot make in place, such as partitioning +an existing BigQuery table or giving it a key), which the gate refuses +without `--allow-data-loss`. On GCP a replaced table comes back with its +governance in the same apply: its policy tags are in the table's schema, +and its row access policies and table-level grants, which BigQuery deletes +with the table, are recreated with it (`replace_triggered_by`; see +[governance parity](governance-parity.md#changes-a-live-table-cannot-take-in-place)). + +`replace-and-build` then runs the builds as a full refresh. dbt gets +`--full-refresh`. An embedded-SQL build into BigQuery lands with one +`WRITE_TRUNCATE_DATA` load into the existing table, which replaces the +rows and keeps the table, its policy tags and its row access policies. +dbt does not: dbt-bigquery materialises a `table` model (and, under +`--full-refresh`, an `incremental` one) with `create or replace table`, +which drops the row access policies and, unless the dbt project re-applies +them, the policy tags `tofu` made. That is a known gap; see +[governance parity](governance-parity.md). + +`fluid rollback` restores a BigQuery table the same way it keeps it: one +query job, `SELECT * FROM ` into the live table with +`WRITE_TRUNCATE_DATA` and `CREATE_NEVER`, never `CREATE OR REPLACE TABLE`. +The restoring identity must read the whole backup: a row access policy +that shows it only some rows makes the rollback refuse +(`rollback_bigquery_backup_filtered`), and a policy-tagged column needs +Fine-Grained Reader on its tag. + **Drift against the apply's own state.** `fluid diff` (and `fluid verify --state-drift`) run the refresh `fluid apply` runs, without applying: from the same workdir and backend (`--workspace-dir`, `--state-backend`, diff --git a/docs/governance-parity.md b/docs/governance-parity.md index 375efcd3..3cc671aa 100644 --- a/docs/governance-parity.md +++ b/docs/governance-parity.md @@ -179,12 +179,31 @@ policy: filter, with the principal's excluded columns as its column wildcard's exclusions, instead of on the table. The masked view of a filtered principal applies the same predicate, since the view reads the table as its definer. -* **A build keeps them.** BigQuery removes every row access policy of a table that a - `WRITE_TRUNCATE` load writes, so a full-refresh build loads with +* **An embedded-SQL build keeps them.** BigQuery removes every row access policy of a + table that a `WRITE_TRUNCATE` load writes, so a full-refresh build loads with `WRITE_TRUNCATE_DATA`, which replaces the rows and keeps the table's policies, tags and partitioning (measured on BigQuery, 5 October 2026, on an unpartitioned and on a DAY-partitioned table with a partition expiration). It is the disposition dlt's BigQuery destination uses for the same reason. +* **A replace and a rollback keep them.** Nothing forge-cli runs recreates a governed + BigQuery table. `fluid rollback` restores one with a single query job, `SELECT * FROM + ` into the live table with `WRITE_TRUNCATE_DATA` and `CREATE_NEVER`; it was + `CREATE OR REPLACE TABLE … AS SELECT`, which BigQuery documents drops every row + access policy of the table, and whose SELECT schema carried no policy tag. It refuses + a backup the restoring identity cannot read in full, which would otherwise restore + only the rows a policy shows it. The GCP planner's replace plans the same query job + for a SQL build, keeps the table definition planned, and backs the table up with a + zero-copy `CREATE SNAPSHOT TABLE … CLONE`, not the plain copy it made, which had no + policy and no tag. +* **Known gap: dbt still recreates the table.** dbt-bigquery materialises a `table` + model with `create or replace table`, in every build mode, and an `incremental` model + the same way under `--mode replace-and-build`, where forge-cli passes `--full-refresh`. + That runs after `tofu apply` made the row access policies and policy tags: the row + access policies are dropped, and the policy tags too unless the dbt project re-applies + them (`persist_docs` with `policy_tags`, which dbt does after the replace). Until the + next `fluid apply`, every principal a filter or a restriction limits reads every row + and column of a table a dbt build recreated. forge-cli does not change dbt's + materialisation yet; that is a follow-up. ## Dataset grants are no longer authoritative @@ -276,6 +295,16 @@ partitioning's shape. Changing `retention` later is an in-place update of also refuses without `--allow-data-loss`; BigQuery cannot un-partition a table, so to keep data longer, set a longer `retention` instead. +A replaced table comes back governed. BigQuery deletes a table's row access policies +and its IAM policy with the table, and `tofu` did not see it in the same apply: a row +access policy's `table_id` is the same string before and after, so it planned the +policies as unchanged and the new table stood with none (and, in a shared dataset, +without its table-level grants) until the next apply. Each row access policy and +table IAM member now has `lifecycle.replace_triggered_by` on the table's `id`, which +is unknown in the plan only when a new table is created, so they are recreated with +the table and left alone by an in-place update. Their replacement adds removals the +data-loss gate counts, but only in a plan that already replaces the table. + ## Prerequisites * The Cloud KMS API (`cloudkms.googleapis.com`) and the Data Catalog API @@ -306,7 +335,10 @@ keep data longer, set a longer `retention` instead. in-process stand-in for the BigQuery REST API, shows adding retention or a key to a live table plans its replacement (the data-loss gate refuses it), a new retention is in place, the governed table then plans clean, and the dataset's own access entries - survive the grants (`tests/iac/test_iac_gcp_governance_plan.py`). + survive the grants (`tests/iac/test_iac_gcp_governance_plan.py`). The stand-in + deletes a table's row access policies and IAM policy with it, as BigQuery does: a + replaced table gets both back in the same apply, and an in-place update (a label, a + new retention) leaves them. * A real `tofu plan` against moto accepts the Lake Formation grants with excluded columns, and `fluid verify`'s Lake Formation check runs against moto's stored grants (`tests/iac/test_iac_aws_column_restrictions.py`). @@ -347,7 +379,10 @@ keep data longer, set a longer `retention` instead. eleven AWS products planned no change for any of them, with only the provider pin moved from `~> 5.0`. * **Not proven yet**: an apply of the masking, row filter and label resources exactly - as forge-cli emits them, on either cloud. + as forge-cli emits them, on either cloud. Nor, on real BigQuery, that a + `WRITE_TRUNCATE_DATA` *query* job (the rollback's and the planner's replace) keeps + the policy tags and row access policies as the measured load job does, how it maps + the SELECT's columns, or what a table snapshot keeps of its table's governance. * **Not proven**: the Lake Formation half against a real account as 0.17.0 derives it from `columnRestrictions`. The grant shape it emits (excluded columns beside `wildcard`) was applied and enforced on a real account from 0.16.6, written by hand diff --git a/fluid_build/cli/_rollback_writer.py b/fluid_build/cli/_rollback_writer.py index 8a2f2bfe..ce2c53d2 100644 --- a/fluid_build/cli/_rollback_writer.py +++ b/fluid_build/cli/_rollback_writer.py @@ -89,10 +89,11 @@ def _restore_ddl_for_snapshot(snapshot_meta: Mapping[str, Any], provider: str) - """Build the DDL ``cli/rollback.py`` will run to restore the snapshot. Delegates to ``provider.restore_ddl(snapshot)`` — every in-tree - provider owns its restore semantics (Snowflake CLONE, BigQuery - CTAS, AWS S3-prefix copy, Redshift transactional CTAS). Adding a - new provider is a single subclass override, not a central - if/elif edit here. + provider owns its restore semantics (Snowflake CLONE, AWS S3-prefix + copy, Redshift transactional CTAS; BigQuery records none, because + ``fluid rollback`` restores it with a data-only query job that keeps + the table and its governance). Adding a new provider is a single + subclass override, not a central if/elif edit here. Returns ``[]`` when no DDL applies — the rollback CLI surfaces that as a typed ``rollback_provider_unsupported`` error. diff --git a/fluid_build/cli/rollback.py b/fluid_build/cli/rollback.py index 1fc6a5f2..637c1c8d 100644 --- a/fluid_build/cli/rollback.py +++ b/fluid_build/cli/rollback.py @@ -30,8 +30,10 @@ Per-provider restore semantics: - **Snowflake** — ``CREATE OR REPLACE DATABASE CLONE `` (zero-copy, reverses the original CLONE used for backup). -- **BigQuery** — NotImplementedError with actionable message. Follow-up - work: ``bq cp --force `` per table. +- **BigQuery** — one query job, ``SELECT * FROM `` into the live + table with ``WRITE_TRUNCATE_DATA`` and ``CREATE_NEVER``: the rows are + replaced, the table and its governance (policy tags, row access policies, + partitioning) are kept. - **Redshift** — NotImplementedError with actionable message. Follow-up work: ``TRUNCATE ; INSERT INTO
SELECT * FROM ;`` inside a transaction. @@ -498,22 +500,44 @@ def _restore_snowflake(snapshot: Dict[str, Any], *, dry_run: bool) -> Dict[str, def _restore_bigquery(snapshot: Dict[str, Any], *, dry_run: bool) -> Dict[str, Any]: - """Restore a BigQuery product via CTAS, reconstructed from validated IDs. - - BigQuery has no zero-copy CLONE for arbitrary backups, so the restore is - ``CREATE OR REPLACE TABLE AS SELECT * FROM `` (atomic, so - the live table is never left half-restored). + """Restore a BigQuery table's rows from its backup, and keep the table. + + The restore is one query job: ``SELECT * FROM `` with the live table + as its destination, ``WRITE_TRUNCATE_DATA`` and ``CREATE_NEVER``. BigQuery + replaces the rows as one atomic step and keeps the existing table: it + "overwrites the data, but keeps the constraints and schema of the existing + table" (REST reference, ``JobConfigurationQuery.writeDisposition``), so the + schema's column policy tags (column-level security and data masking) stay as + ``tofu apply`` made them, and it is not the ``WRITE_TRUNCATE`` BigQuery + documents removes every row access policy (a ``WRITE_TRUNCATE_DATA`` load kept + them, measured on BigQuery on 5 October 2026). It was ``CREATE OR REPLACE + TABLE AS SELECT * FROM ``, and BigQuery documents that + replacing a table that way drops all of its row access policies; the new + table took its schema from the SELECT, so the policy tags went too, and every + filtered or restricted reader read every row and column until the next + apply. The disposition and the client are the load path's + (``build_runners/_bigquery_load.py``, after dlt's BigQuery destination): + ``WRITE_TRUNCATE`` on an emulator, which has no policy to keep. + + The backup is read as the restoring identity. A row access policy that + filters it for that identity would restore a subset and delete the rest, so + the rows it can read are counted against the backup's own row count first, + and a restore that would lose rows is refused before anything is written. + Reading a column with a policy tag needs Fine-Grained Reader on the tag; + without it the job fails and the table is left as it was. SECURITY: the on-disk state file (``.fluid/rollback-state.json``) is attacker-authorable — the module docstring invites operators to commit it as an audit trail, so it is reviewed as a data blob, not byte-audited SQL. - The baked ``snapshot["ddl"]`` is therefore NEVER executed: the statement - is REBUILT here from ``location`` components that each pass + The baked ``snapshot["ddl"]`` is therefore NEVER executed: the query is + REBUILT here from ``location`` components that each pass ``_validated_bq_fqn``, so a tampered ddl[] (e.g. ``DROP TABLE prod``) - cannot smuggle arbitrary DDL past a benign-looking location. Mirrors the - apply-time writer ``GcpProvider.restore_ddl``. Returns the usual - ``{status, provider, ddl, ...}`` contract. + cannot smuggle arbitrary SQL past a benign-looking location, and the + destination is a ``TableReference`` of the same validated identifiers, + never SQL. Returns the usual ``{status, provider, ddl, ...}`` contract, + ``ddl`` being the query the restore runs. """ + from fluid_build.build_runners import _bigquery_load from fluid_build.providers.gcp.plan.planner import _validated_bq_fqn location = snapshot.get("location") or {} @@ -547,13 +571,13 @@ def _restore_bigquery(snapshot: Dict[str, Any], *, dry_run: bool) -> Dict[str, A ), }, ) - # SECURITY: reconstruct the restore DDL from the VALIDATED identifiers - # instead of replaying snapshot["ddl"] verbatim. The state file is - # attacker-authorable, so a tampered ddl[] with a benign location could - # otherwise execute arbitrary SQL (DROP/DELETE/GRANT) under the operator's - # credentials. We ONLY ever run the single, structurally-safe - # ``CREATE OR REPLACE TABLE AS SELECT * FROM `` built from - # FQNs that passed _validated_bq_fqn; the baked ddl[] is never executed. + # SECURITY: reconstruct the restore from the VALIDATED identifiers instead + # of replaying snapshot["ddl"] verbatim. The state file is attacker- + # authorable, so a tampered ddl[] with a benign location could otherwise + # execute arbitrary SQL (DROP/DELETE/GRANT) under the operator's + # credentials. We ONLY ever run ``SELECT * FROM `` built from an FQN + # that passed _validated_bq_fqn, into a destination built from identifiers + # that passed it too; the baked ddl[] is never executed. try: orig_fqn = _validated_bq_fqn(raw_project, raw_dataset, raw_table) backup_fqn = _validated_bq_fqn(raw_project, raw_dataset, raw_backup) @@ -572,21 +596,42 @@ def _restore_bigquery(snapshot: Dict[str, Any], *, dry_run: bool) -> Dict[str, A }, ) from exc - ddl_statements = [f"CREATE OR REPLACE TABLE {orig_fqn} AS SELECT * FROM {backup_fqn}"] - ddl = ddl_statements[0] + ";" + query = f"SELECT * FROM {backup_fqn}" + write_disposition = _bigquery_load.write_disposition("full_refresh") + create_disposition = "CREATE_NEVER" + plan = { + "provider": "bigquery", + "ddl": query, + "destination": orig_fqn, + "write_disposition": write_disposition, + "create_disposition": create_disposition, + "project": raw_project, + "backup_name": raw_backup, + } cprint( - "[rollback] bigquery CTAS restore plan:\n " - + "\n ".join(stmt + ";" for stmt in ddl_statements), + "[rollback] bigquery restore plan (rows only; the table, its policy tags " + "and its row access policies are kept):\n" + f" {query};\n" + f" destination {orig_fqn}, {write_disposition}, {create_disposition}", markup=False, ) if dry_run: - return { - "status": "dry_run", - "provider": "bigquery", - "ddl": ddl, - "project": raw_project, - "backup_name": raw_backup, - } + return {"status": "dry_run", **plan} + validated = { + "project": raw_project, + "dataset": raw_dataset, + "table": raw_table, + "backup": raw_backup, + "backup_fqn": backup_fqn, + } + provider_result = _run_bigquery_restore(validated, plan) + return {"status": "restored", **plan, "provider_result": provider_result} + + +def _run_bigquery_restore(ids: Dict[str, str], plan: Dict[str, Any]) -> Dict[str, Any]: + """Run the restore :func:`_restore_bigquery` planned, from the identifiers it validated.""" + from fluid_build.build_runners import _bigquery_load + # Execute via the BigQuery client. Imported lazily so non-BigQuery # rollbacks don't pay the google-cloud-bigquery import cost (mirrors # the lazy SnowflakeProvider import in _restore_snowflake and the @@ -607,27 +652,72 @@ def _restore_bigquery(snapshot: Dict[str, Any], *, dry_run: bool) -> Dict[str, A }, ) from exc - client = bigquery.Client(project=raw_project) - last_result: Any = None - for stmt in ddl_statements: - try: - job = client.query(stmt) - last_result = job.result() # blocks until the statement completes - except Exception as exc: # pragma: no cover - defensive - raise CLIError( - 2, - "rollback_bigquery_execute_failed", - {"error": str(exc), "ddl": stmt}, - ) from exc + client = _bigquery_load.bigquery_client(bigquery, ids["project"]) + dataset_ref = bigquery.DatasetReference(ids["project"], ids["dataset"]) + readable: Optional[int] = None + if _bigquery_load.emulator_host() is None: + readable = _readable_backup_rows( + client, + bigquery.TableReference(dataset_ref, ids["backup"]), + f"{ids['project']}.{ids['dataset']}.{ids['backup']}", + ids["backup_fqn"], + ) + query = plan["ddl"] + job_config = bigquery.QueryJobConfig( + destination=bigquery.TableReference(dataset_ref, ids["table"]), + write_disposition=plan["write_disposition"], + create_disposition=plan["create_disposition"], + ) + try: + job = client.query(query, job_config=job_config) + job.result() # blocks until the job completes + except Exception as exc: # pragma: no cover - defensive + raise CLIError( + 2, + "rollback_bigquery_execute_failed", + {"error": str(exc), "ddl": query, "destination": plan["destination"]}, + ) from exc + return {"status": "ok", "job_id": getattr(job, "job_id", None), "rows": readable} - return { - "status": "restored", - "provider": "bigquery", - "ddl": ddl, - "project": raw_project, - "backup_name": raw_backup, - "provider_result": {"status": "ok", "statements": len(ddl_statements)}, - } + +def _readable_backup_rows(client: Any, backup_ref: Any, backup_id: str, backup_fqn: str) -> int: + """The backup's rows, refused unless the restoring identity can read every one. + + A row access policy on the backup (BigQuery supports row-level security on + table snapshots) shows the identity only the rows it is granted, and + ``WRITE_TRUNCATE_DATA`` would then replace the table with them. ``num_rows`` + is the table's own count, which no policy filters. + """ + from fluid_build.build_runners import _bigquery_load + + try: + stored = client.get_table(backup_ref).num_rows + readable = _bigquery_load._count_table(client, backup_id, None) + except Exception as exc: + raise CLIError( + 2, + "rollback_bigquery_execute_failed", + {"error": str(exc), "ddl": f"SELECT COUNT(*) FROM {backup_fqn}"}, + ) from exc + if stored is None or readable < int(stored): + raise CLIError( + 2, + "rollback_bigquery_backup_filtered", + { + "backup": backup_fqn, + "rows": stored, + "readable": readable, + "hint": ( + "the identity running the rollback cannot read every row of the " + "backup (a row access policy filters it), so restoring would " + "replace the table with only those rows. Nothing was written. " + "Run the rollback as an identity a TRUE row access policy covers " + "(fluid apply's fluid_all_rows covers each reader and writer of " + "the expose that no row filter names)." + ), + }, + ) + return readable def _restore_redshift(snapshot: Dict[str, Any], *, dry_run: bool) -> Dict[str, Any]: diff --git a/fluid_build/cli/scaffold_ide.py b/fluid_build/cli/scaffold_ide.py index b8dd5f82..0ed2a88b 100644 --- a/fluid_build/cli/scaffold_ide.py +++ b/fluid_build/cli/scaffold_ide.py @@ -186,6 +186,13 @@ def register(subparsers: argparse._SubParsersAction): `--allow-data-loss` is required for `replace*` when `FLUID_ENV != dev` OR target has rows. Never bypass without explicit user confirmation. +The `replace` rows are the native engine's. On the OpenTofu engine (aws, gcp, +snowflake, confluent) `replace` applies the same module as `amend` behind the +data-loss gate. A BigQuery table is never recreated to replace its rows: a +replace or a `fluid rollback` writes them with a `WRITE_TRUNCATE_DATA` query +job, which keeps the table's policy tags and row access policies. dbt's `table` +materialisation still recreates the table. See `docs/apply.md`. + ## Plan-binding (Terraform-style "apply consumes exact plan") `fluid plan` emits two cryptographic fields into `plan.json`: diff --git a/fluid_build/providers/_planner_base.py b/fluid_build/providers/_planner_base.py index 62ab5fcf..08898e94 100644 --- a/fluid_build/providers/_planner_base.py +++ b/fluid_build/providers/_planner_base.py @@ -19,14 +19,19 @@ 1. **Infrastructure** — datasets / databases / schemas / buckets. 2. **IAM** — roles, grants, masking + row-access policies. -3. **Replace snapshots** — pre-flight backups (CLONE / CTAS / S3 copy) - emitted only for destructive modes (``replace`` / ``replace-and-build``). +3. **Replace snapshots** — pre-flight backups (Snowflake CLONE, BigQuery + table snapshot, S3 copy) emitted only for destructive modes + (``replace`` / ``replace-and-build``). 4. **Expose** — the actual data products (tables / views / streams) the contract publishes. Runs BEFORE build so SQL transforms have - their target table to write into (additive ``INSERT INTO`` mode). + their target table to write into. 5. **Build** — transformations: dbt / Dataform / SQL / stored procedures / - tasks / UDFs. Destructive mode emits ``CREATE OR REPLACE TABLE … - AS SELECT``; additive emits ``INSERT INTO``. + tasks / UDFs. Additive mode emits ``INSERT INTO``. Destructive mode + replaces the target's rows, each provider its own way: Snowflake emits + ``CREATE OR REPLACE TABLE … AS SELECT``; GCP emits the SELECT as a + ``WRITE_TRUNCATE_DATA`` query job into the existing table, because + recreating a BigQuery table drops its row access policies and policy + tags. 6. **Schedule** — task orchestration, pipes, Cloud Scheduler / Composer. Before this module the scaffold lived as a copy-pasted ``plan_actions`` @@ -114,8 +119,9 @@ def plan( Phase 3 (replace_snapshots) is skipped for additive modes. Phases 4 (expose) + 5 (build) receive the ``is_destructive`` - flag so they can route between ``CREATE OR REPLACE TABLE … - AS SELECT`` (destructive) and ``INSERT INTO`` (additive). + flag so they can route between a replace of the target's rows + (destructive; see the module docstring for each provider's) and + ``INSERT INTO`` (additive). """ is_destructive = is_destructive_mode(mode) actions: List[Dict[str, Any]] = [] @@ -157,8 +163,9 @@ def plan_expose( ) -> List[Dict[str, Any]]: """Phase 4 — exposed data products (tables, views, streams). - ``is_destructive=True`` typically suppresses the ensure_table - action because Phase 5's CREATE OR REPLACE handles materialisation. + ``is_destructive=True`` lets a provider whose Phase 5 recreates the + table (Snowflake's ``CREATE OR REPLACE``) skip its ensure_table. One + that replaces only the rows (GCP) keeps the table planned. """ return [] @@ -170,8 +177,9 @@ def plan_build( ) -> List[Dict[str, Any]]: """Phase 5 — SQL transformations (dbt / Dataform / stored procs). - ``is_destructive=True`` switches from ``INSERT INTO`` to - ``CREATE OR REPLACE TABLE … AS SELECT``. + ``is_destructive=True`` switches from ``INSERT INTO`` to a replace + of the target's rows (Snowflake: ``CREATE OR REPLACE TABLE … AS + SELECT``; GCP: a ``WRITE_TRUNCATE_DATA`` query job). """ return [] diff --git a/fluid_build/providers/gcp/plan/planner.py b/fluid_build/providers/gcp/plan/planner.py index 19c41224..64f70c50 100644 --- a/fluid_build/providers/gcp/plan/planner.py +++ b/fluid_build/providers/gcp/plan/planner.py @@ -44,8 +44,9 @@ def _validate_bq_project(project: str) -> str: Returns the project ID unchanged when valid; raises ``ValueError`` otherwise. ``project`` flows into a backtick-quoted BigQuery FQN in - CTAS / INSERT DDL — backticked or not, an unvetted value is an - injection vector, so the project component is allowlisted by shape. + snapshot / INSERT / SELECT statements — backticked or not, an unvetted + value is an injection vector, so the project component is allowlisted + by shape. """ if not isinstance(project, str) or not _GCP_PROJECT_RE.match(project): raise ValueError(f"Invalid GCP project ID: {project!r}") @@ -130,24 +131,27 @@ def plan_actions( Generate GCP actions from FLUID contract. Back-compat shim — constructs a :class:`GcpPlanner` and calls - :meth:`GcpPlanner.plan`. The 6-phase ordering, destructive-mode - handling, and CTAS backup emission live in + :meth:`GcpPlanner.plan`. The 6-phase ordering and destructive-mode + routing live in :class:`fluid_build.providers._planner_base.BasePlanner` so the GCP provider stays in lockstep with Snowflake / AWS. Phase order (set by ``BasePlanner.plan``): 1. Infrastructure — datasets and buckets. 2. IAM — service-account bindings, custom roles. - 3. Replace snapshots — pre-flight CTAS per bigquery_table - (destructive modes only). + 3. Replace snapshots — pre-flight ``CREATE SNAPSHOT TABLE`` per + bigquery_table (destructive modes only). 4. Expose — tables, APIs, streams. 5. Build — dbt / Dataform / SQL transforms. 6. Schedule — Composer DAGs / Cloud Scheduler. - For destructive modes (``replace`` / ``replace-and-build``), - SQL builds emit ``CREATE OR REPLACE TABLE … AS SELECT`` (BigQuery's - atomic replace) and a pre-flight CTAS backup is emitted per - ``bigquery_table`` expose for rollback. + For destructive modes (``replace`` / ``replace-and-build``), a SQL + build replaces only its target's rows: the action carries the SELECT, + a structured ``destination`` and ``WRITE_TRUNCATE_DATA`` / + ``CREATE_NEVER`` for a query job, never ``CREATE OR REPLACE TABLE``, + which drops the table's row access policies and policy tags. The + table stays planned (``bq.ensure_table``), and a zero-copy snapshot + of each ``bigquery_table`` expose is taken first for rollback. Args: contract: FLUID contract specification @@ -391,8 +395,9 @@ def _plan_build_transformations( Sets up dbt, Dataform, or other transformation engines. When ``is_destructive`` is True, the action is stamped with - ``mode="replace"`` so the dbt executor adds ``--full-refresh`` - and SQL transforms emit ``CREATE OR REPLACE TABLE … AS SELECT``. + ``mode="replace"`` so the dbt executor adds ``--full-refresh``, + and a SQL transform into a BigQuery table replaces that table's rows + (:func:`_bq_build_for_target`), keeping the table and its governance. """ actions = [] apply_mode = "replace" if is_destructive else "amend" @@ -419,9 +424,10 @@ def _plan_build_transformations( # Inline-SQL build path (mirrors the Snowflake planner's # ``_plan_build`` SQL emission). For each ``builds[]`` entry with # ``properties.sql`` AND outputs targeting a ``bigquery_table`` - # expose, emit ``bq.sql.execute`` actions wrapping the SQL into - # INSERT INTO (additive) or CREATE OR REPLACE TABLE … AS SELECT - # (destructive). Multi-output builds emit one action per output. + # expose, emit ``bq.sql.execute`` actions that write the SQL into it: + # INSERT INTO (additive), or the SELECT as a query job that replaces + # the table's rows (destructive). Multi-output builds emit one action + # per output. for build_idx, build_entry in enumerate(contract.get("builds", []) or []): if not isinstance(build_entry, Mapping): continue @@ -436,8 +442,8 @@ def _plan_build_transformations( outputs = [None] build_id = build_entry.get("id", f"build_{build_idx}") for out_id in outputs: - wrapped = ( - _bq_wrap_sql_for_target( + statement = ( + _bq_build_for_target( sql_text=sql_text, contract=contract, target_output_id=out_id, @@ -445,7 +451,7 @@ def _plan_build_transformations( is_destructive=is_destructive, ) if out_id - else sql_text + else {"sql": sql_text} ) action_id = build_id if len(outputs) == 1 else f"{build_id}__{out_id}" actions.append( @@ -454,7 +460,7 @@ def _plan_build_transformations( "op": "bq.sql.execute", "phase": "build", "project": project, - "sql": wrapped, + **statement, "mode": apply_mode, "comment": build_entry.get("description") or build_entry.get("name"), } @@ -465,21 +471,11 @@ def _plan_build_transformations( return actions -def _bq_wrap_sql_for_target( - *, - sql_text: str, - contract: Mapping[str, Any], - target_output_id: str, - default_project: str, - is_destructive: bool, -) -> str: - """Wrap a build's SQL with INSERT or CTAS for a BigQuery target. - - Mirrors :func:`fluid_build.providers.snowflake.plan.planner._wrap_sql_for_target` - using BigQuery's backtick-quoted identifier syntax. Pass-through - when the target expose isn't a ``bigquery_table`` binding or when - the SQL already declares its own sink (INSERT/CREATE/MERGE/etc). - """ +def _bq_build_target( + contract: Mapping[str, Any], target_output_id: str, default_project: str +) -> Optional[tuple]: + """``(project, dataset, table)`` of the ``bigquery_table`` expose a build output + names, or None when it names none (or one without a dataset).""" target_expose = next( ( ex @@ -490,29 +486,88 @@ def _bq_wrap_sql_for_target( None, ) if target_expose is None: - return sql_text + return None binding = target_expose.get("binding") or {} if (binding.get("format") or "").lower() not in ( "bigquery_table", "bigquery-table", ): - return sql_text + return None location = binding.get("location") or {} proj = location.get("project") or default_project dataset = location.get("dataset") or location.get("schema") table = location.get("table") or target_output_id if not (proj and dataset and table): - return sql_text + return None + return proj, dataset, table + + +#: How a destructive SQL build writes its BigQuery target: a query job that +#: replaces the rows of the table ``tofu apply`` owns, never creates one +#: (``build_runners/_bigquery_load.py`` loads with the same two). +_REPLACE_ROWS_DISPOSITIONS = { + "write_disposition": "WRITE_TRUNCATE_DATA", + "create_disposition": "CREATE_NEVER", +} + + +def _bq_build_for_target( + *, + sql_text: str, + contract: Mapping[str, Any], + target_output_id: str, + default_project: str, + is_destructive: bool, +) -> Dict[str, Any]: + """The action fields that write a build's SQL into its BigQuery target. + + Additive: ``{"sql": "INSERT INTO "}``. Destructive: the SELECT + itself, the target as a structured ``destination`` (``project``, + ``dataset``, ``table``, each validated) and the query job's + ``WRITE_TRUNCATE_DATA`` / ``CREATE_NEVER``. BigQuery then replaces the + rows of the existing table and keeps the table: it "overwrites the data, + but keeps the constraints and schema of the existing table" (REST + reference, ``JobConfigurationQuery.writeDisposition``), so the schema's + policy tags stay, and the row access policies, which ``CREATE OR REPLACE + TABLE … AS SELECT`` (what this emitted) drops, are not touched. That is + the dlt BigQuery destination's atomic replace, extended from a load job to + a query job; the table and its governance stay the IaC's. + + Mirrors :func:`fluid_build.providers.snowflake.plan.planner._wrap_sql_for_target` + using BigQuery's backtick-quoted identifier syntax. Pass-through + (``{"sql": sql_text}``, untouched) when the target expose isn't a + ``bigquery_table`` binding or when the SQL already declares its own sink + (INSERT/CREATE/MERGE/etc). + """ + target = _bq_build_target(contract, target_output_id, default_project) + if target is None: + return {"sql": sql_text} + proj, dataset, table = target upper_head = sql_text.lstrip().upper()[:32] for kw in ("INSERT", "CREATE", "MERGE", "UPDATE", "DELETE", "COPY", "TRUNCATE"): if upper_head.startswith(kw): - return sql_text + return {"sql": sql_text} body = sql_text.rstrip().rstrip(";") - # Validate every component of the FQN before it lands in DDL/DML. + # Validate every component of the FQN before it lands in DML or in the + # destination a query job is given. fqn = _validated_bq_fqn(proj, dataset, table) if is_destructive: - return f"CREATE OR REPLACE TABLE {fqn} AS\n{body}" - return f"INSERT INTO {fqn}\n{body}" + return { + "sql": body, + "destination": {"project": proj, "dataset": dataset, "table": table}, + **_REPLACE_ROWS_DISPOSITIONS, + } + return {"sql": f"INSERT INTO {fqn}\n{body}"} + + +#: How long a pre-replace table snapshot is kept. The copy it replaces had no +#: expiry and was deleted only when the rollback state pruned it +#: (``cli/_rollback_writer.py``, the last 20 per env and product); a snapshot +#: is a table to ``GcpProvider.cleanup_backups`` too, so the pruning still +#: deletes it, and the expiry bounds one the pruning never reaches. There is no +#: GCP setting for it; thirty days is the backup retention the Snowflake +#: provider's configuration defaults to (``BackupConfig.backup_retention``). +_SNAPSHOT_RETENTION_DAYS = 30 def _plan_replace_snapshots( @@ -521,23 +576,25 @@ def _plan_replace_snapshots( region: str, logger: logging.Logger, ) -> List[Dict[str, Any]]: - """Pre-flight CTAS backups for ``--mode replace`` against BigQuery. + """Pre-flight table snapshots for ``--mode replace`` against BigQuery. For each ``bigquery_table`` expose, emit an action that runs - ``CREATE TABLE IF NOT EXISTS AS SELECT * FROM `` - BEFORE the destructive replace fires. BigQuery has no CLONE so the - backup is a real copy (storage cost applies); the - ``IF NOT EXISTS`` guard keeps re-runs idempotent. - - The action carries a ``rollback_snapshot`` marker so apply.py's - ``_rollback_writer`` can record the snapshot in - ``.fluid/rollback-state.json`` for ``fluid rollback``. - - Skips exposes whose source table doesn't exist yet (first-run - replace) — the IF NOT EXISTS gate also covers the pre-existence - case but BigQuery raises if the SOURCE is absent. The backup - action is marked ``allow_failure: True`` so a missing source on - first-run replace soft-skips rather than aborting the plan. + ``CREATE SNAPSHOT TABLE IF NOT EXISTS CLONE + OPTIONS(expiration_timestamp = ...)`` BEFORE the destructive replace + fires: BigQuery's zero-copy, read-only backup, billed only for the data + that later changes in the base table ("Introduction to table + snapshots"). It was ``CREATE TABLE … AS SELECT * FROM ``, a full + copy with no row access policy and no policy tag, readable in full by + every reader of the dataset, justified here by "BigQuery has no CLONE" + (it has). It expires after :data:`_SNAPSHOT_RETENTION_DAYS`, or when + the rollback-state pruning drops it; ``fluid rollback`` restores from + it with a data-only query job. ``IF NOT EXISTS`` keeps re-runs + idempotent. + + The ``rollback_snapshot`` marker lets ``_rollback_writer`` record the + snapshot in ``.fluid/rollback-state.json``. BigQuery raises when the + SOURCE does not exist yet (first-run replace), so the action is + ``allow_failure: True`` and soft-skips rather than aborting the plan. """ import time as _time @@ -560,10 +617,14 @@ def _plan_replace_snapshots( continue backup = f"BACKUP_{table}_{backup_ts}" # Validate every FQN component (project shape + dataset/table - # identifiers) before the CTAS backup statement is emitted. + # identifiers) before the snapshot statement is emitted. backup_fqn = _validated_bq_fqn(proj, dataset, backup) source_fqn = _validated_bq_fqn(proj, dataset, table) - sql = f"CREATE TABLE IF NOT EXISTS {backup_fqn} AS SELECT * FROM {source_fqn}" + sql = ( + f"CREATE SNAPSHOT TABLE IF NOT EXISTS {backup_fqn} CLONE {source_fqn} " + "OPTIONS(expiration_timestamp = TIMESTAMP_ADD(CURRENT_TIMESTAMP(), " + f"INTERVAL {_SNAPSHOT_RETENTION_DAYS} DAY))" + ) actions.append( { "id": f"snapshot_{expose.get('exposeId')}", @@ -601,34 +662,18 @@ def _plan_exposures( """ Plan data product exposure actions. - Creates tables, views, APIs, streams, etc. When ``is_destructive`` - is True and the expose is a target of a SQL build (listed in - ``builds[].outputs``), the ensure_table step is skipped because - the build's CREATE OR REPLACE TABLE handles materialisation. + Creates tables, views, APIs, streams, etc. ``is_destructive`` changes + nothing here: a destructive SQL build replaces only its target's rows + (:func:`_bq_build_for_target`), so the table, with the policy tags in + its schema, stays planned in every mode. It used to be skipped for a + SQL-build target, left to a ``CREATE OR REPLACE TABLE`` that dropped + the table's governance. """ actions = [] - # Build the set of expose ids targeted by SQL builds; their - # ensure_table is skipped under destructive modes. - sql_build_targets: set = set() - if is_destructive: - for build_entry in contract.get("builds", []) or []: - if not isinstance(build_entry, Mapping): - continue - outputs = build_entry.get("outputs") or [] - if outputs and ( - build_entry.get("sql") or (build_entry.get("properties") or {}).get("sql") - ): - sql_build_targets.update(outputs) - for exposure in contract.get("exposes", []): exposure_id = exposure.get("id") or exposure.get("exposeId") exposure.get("type") or exposure.get("kind") - # Skip ensure_table for destructive-mode SQL-build targets — - # CREATE OR REPLACE TABLE in the build phase materialises them. - if is_destructive and exposure_id in sql_build_targets: - continue - # Support both old and new structures location = exposure.get("location", {}) binding = exposure.get("binding", {}) diff --git a/fluid_build/providers/gcp/provider.py b/fluid_build/providers/gcp/provider.py index 3ebb00fc..185895b5 100644 --- a/fluid_build/providers/gcp/provider.py +++ b/fluid_build/providers/gcp/provider.py @@ -96,34 +96,19 @@ def capabilities(self) -> Mapping[str, bool]: } def restore_ddl(self, snapshot: Mapping[str, Any]) -> List[str]: - """BigQuery restore via CTAS (BigQuery has no CLONE). - - ``CREATE OR REPLACE TABLE AS SELECT * FROM `` - is atomic. Storage cost applies (the backup is a real copy - of the data, not a metadata pointer). - - Every component of both fully-qualified names is validated - before interpolation: the project component via the GCP - project-ID shape check, and the dataset / table components via - ``_sql_safety.validate_ident``. Invalid snapshot metadata yields - an empty DDL list rather than an unsafe statement. + """BigQuery has no restore DDL to record: returns ``[]``, as AWS does. + + ``fluid rollback`` restores a BigQuery table with a data-only query job + (``cli/rollback.py::_restore_bigquery``): ``SELECT * FROM `` + into the live table with ``WRITE_TRUNCATE_DATA`` and ``CREATE_NEVER``, + which replaces the rows and keeps the table, its column policy tags and + its row access policies. No SQL statement does that. The recipe this + recorded was ``CREATE OR REPLACE TABLE AS SELECT * FROM + ``, which drops every row access policy and every policy tag of + the table, and the rollback never ran the recorded ``ddl[]`` anyway: it + rebuilds the job from the validated ``location``. """ - from .plan.planner import _validated_bq_fqn - - location = snapshot.get("location") or {} - db = location.get("database") - sch = location.get("schema") - tbl = location.get("table") - backup = location.get("backup_table") or snapshot.get("backup_name") - if not (db and sch and tbl and backup): - return [] - try: - orig_fqn = _validated_bq_fqn(db, sch, tbl) - backup_fqn = _validated_bq_fqn(db, sch, backup) - except ValueError as exc: - self.warn_kv(event="restore_ddl_invalid_identifier", error=str(exc)) - return [] - return [f"CREATE OR REPLACE TABLE {orig_fqn} AS SELECT * FROM {backup_fqn}"] + return [] def cleanup_backups(self, snapshots: List[Mapping[str, Any]]) -> None: """Drop BigQuery backup tables (best-effort).""" @@ -167,10 +152,13 @@ def plan( - IAM policy bindings The optional ``mode`` argument carries the apply-time mode. - For destructive modes (``replace`` / ``replace-and-build``) - the BigQuery actions emit ``CREATE OR REPLACE TABLE … AS SELECT`` - instead of additive ``INSERT INTO``, with a pre-flight backup - CTAS for rollback. + For destructive modes (``replace`` / ``replace-and-build``) a SQL + build replaces only its target's rows (the SELECT, a structured + destination and ``WRITE_TRUNCATE_DATA``) instead of the additive + ``INSERT INTO``, the table definition stays planned, and a + pre-flight ``CREATE SNAPSHOT TABLE`` backs each table up for + rollback. Nothing recreates the table, so its policy tags and row + access policies are kept. """ self.debug_kv( event="plan_started", diff --git a/tests/cli/test_rollback.py b/tests/cli/test_rollback.py index 462b9e0e..20f862f3 100644 --- a/tests/cli/test_rollback.py +++ b/tests/cli/test_rollback.py @@ -509,10 +509,11 @@ def _bq_snap(*, provider="bigquery", backup_name="subscriber360_bak_1", **loc_ov """Build a canonical BigQuery rollback snapshot. Mirrors what providers/gcp/plan/planner.py emits: location carries the - project/dataset/table/backup_table, and ``ddl[]`` is the pre-baked CTAS - restore statement (``CREATE OR REPLACE TABLE AS SELECT * FROM - ``). ``provider`` defaults to "bigquery" but a REAL apply records - "gcp" (GcpProvider.name) — pass provider="gcp" to exercise the dispatch. + project/dataset/table/backup_table. ``ddl[]`` is what an older forge-cli + recorded (``CREATE OR REPLACE TABLE AS SELECT * FROM ``); + the restore never runs it. ``provider`` defaults to "bigquery" but a REAL + apply records "gcp" (GcpProvider.name) — pass provider="gcp" to exercise + the dispatch. """ location = { "database": "myproj", @@ -537,14 +538,58 @@ def _bq_snap(*, provider="bigquery", backup_name="subscriber360_bak_1", **loc_ov } +_BQ_RESTORE_QUERY = "SELECT * FROM `myproj.telco.subscriber360_bak_1`" + + +def _mock_bigquery(*, rows=3, readable=3): + """A ``google.cloud.bigquery`` stand-in injected through ``sys.modules``. + + The backup holds ``rows`` rows (``Table.num_rows``, which no row access + policy filters) and its ``COUNT(*)`` reads ``readable`` of them. Hermetic + whether or not google-cloud-bigquery is installed. + """ + mock_bq = MagicMock() + mock_client = MagicMock() + mock_bq.Client.return_value = mock_client + mock_client.get_table.return_value.num_rows = rows + + def query(sql, **_kwargs): + job = MagicMock() + job.job_id = "job_restore" + job.result.return_value = [{"f0_": readable}] if "COUNT(*)" in sql else [] + return job + + mock_client.query.side_effect = query + google_cloud_mod = MagicMock() + google_cloud_mod.bigquery = mock_bq + modules = { + "google": MagicMock(), + "google.cloud": google_cloud_mod, + "google.cloud.bigquery": mock_bq, + } + return patch.dict(sys.modules, modules), mock_bq, mock_client + + +@pytest.fixture +def _no_emulator(monkeypatch): + monkeypatch.delenv("BIGQUERY_EMULATOR_HOST", raising=False) + + +@pytest.mark.usefixtures("_no_emulator") class TestRestoreBigQuery: - def test_dry_run_returns_ddl_without_client(self): - """Dry-run returns the baked CTAS DDL and never constructs a client.""" - result = rollback._restore_bigquery(_bq_snap(), dry_run=True) + def test_dry_run_returns_the_query_job_without_client(self): + """Dry-run returns the data-only restore and never constructs a client.""" + modules, mock_bq, _ = _mock_bigquery() + with modules: + result = rollback._restore_bigquery(_bq_snap(), dry_run=True) + mock_bq.Client.assert_not_called() assert result["status"] == "dry_run" assert result["provider"] == "bigquery" - assert "CREATE OR REPLACE TABLE" in result["ddl"] - assert _bq_snap()["backup_name"] in result["ddl"] + assert result["ddl"] == _BQ_RESTORE_QUERY + assert result["destination"] == "`myproj.telco.subscriber360`" + assert result["write_disposition"] == "WRITE_TRUNCATE_DATA" + assert result["create_disposition"] == "CREATE_NEVER" + assert "CREATE OR REPLACE" not in json.dumps(result) def test_missing_project_raises(self): snap = _bq_snap() @@ -562,80 +607,119 @@ def test_missing_location_fields_raises(self): rollback._restore_bigquery(snap, dry_run=False) def test_missing_ddl_is_ignored_and_reconstructed(self): - """SECURITY: the baked ddl[] is no longer trusted. A snapshot without - it still restores, because the CTAS is reconstructed from the - validated location identifiers.""" + """SECURITY: the baked ddl[] is not trusted. A snapshot without it still + restores, because the query is rebuilt from the validated location.""" snap = _bq_snap() snap.pop("ddl") result = rollback._restore_bigquery(snap, dry_run=True) assert result["status"] == "dry_run" - assert "CREATE OR REPLACE TABLE" in result["ddl"] - assert "AS SELECT * FROM" in result["ddl"] + assert result["ddl"] == _BQ_RESTORE_QUERY def test_malicious_ddl_is_not_executed(self): """SECURITY (regression for the verbatim-ddl[]-replay finding): a tampered ddl[] with a benign location must NOT execute the attacker's - SQL — only the safe CTAS rebuilt from validated identifiers runs.""" + SQL — only the restore rebuilt from validated identifiers runs.""" snap = _bq_snap() snap["ddl"] = [ "DROP TABLE `myproj.prod.billing`", "DELETE FROM `myproj.prod.users`", ] - mock_bq = MagicMock() - mock_client = MagicMock() - mock_bq.Client.return_value = mock_client - google_cloud_mod = MagicMock() - google_cloud_mod.bigquery = mock_bq - with patch.dict( - sys.modules, - { - "google": MagicMock(), - "google.cloud": google_cloud_mod, - "google.cloud.bigquery": mock_bq, - }, - ): + modules, _, mock_client = _mock_bigquery() + with modules: result = rollback._restore_bigquery(snap, dry_run=False) assert result["status"] == "restored" executed = [c.args[0] for c in mock_client.query.call_args_list] - assert len(executed) == 1 - assert executed[0].startswith("CREATE OR REPLACE TABLE") - assert "AS SELECT * FROM" in executed[0] + assert executed == [ + "SELECT COUNT(*) FROM `myproj`.`telco`.`subscriber360_bak_1`", + _BQ_RESTORE_QUERY, + ] assert all("DROP" not in s and "DELETE" not in s for s in executed) - def test_invalid_identifier_refused(self): - """The state file is attacker-authorable; a tampered dataset with - SQL metacharacters must be refused before any DDL executes.""" - snap = _bq_snap(schema="ds; DROP TABLE x") - with pytest.raises(CLIError, match="rollback_bigquery_invalid_identifier"): - rollback._restore_bigquery(snap, dry_run=False) - - def test_live_restore_invokes_client(self): - """Non-dry-run constructs bigquery.Client(project=...) and runs each - baked statement through it. The SDK is injected via sys.modules so the - test is hermetic regardless of whether google-cloud-bigquery is installed. - """ - snap = _bq_snap() - mock_bq = MagicMock() - mock_client = MagicMock() - mock_bq.Client.return_value = mock_client - google_cloud_mod = MagicMock() - google_cloud_mod.bigquery = mock_bq - with patch.dict( - sys.modules, - { - "google": MagicMock(), - "google.cloud": google_cloud_mod, - "google.cloud.bigquery": mock_bq, - }, - ): - result = rollback._restore_bigquery(snap, dry_run=False) + @pytest.mark.parametrize( + "field, value", + [ + ("schema", "ds; DROP TABLE x"), + ("table", "t` ; DROP TABLE x; --"), + ("backup_table", "b`; GRANT ALL"), + ("database", "Proj With Spaces"), + ], + ) + def test_invalid_identifier_refused_before_any_client(self, field, value): + """The state file is attacker-authorable; a tampered location value with + SQL metacharacters is refused before a client exists or a job runs.""" + modules, mock_bq, mock_client = _mock_bigquery() + with modules: + with pytest.raises(CLIError, match="rollback_bigquery_invalid_identifier"): + rollback._restore_bigquery(_bq_snap(**{field: value}), dry_run=False) + mock_bq.Client.assert_not_called() + mock_client.query.assert_not_called() + + def test_live_restore_replaces_the_rows_and_keeps_the_table(self): + """One query job into the live table, ``WRITE_TRUNCATE_DATA`` and + ``CREATE_NEVER``: never ``CREATE OR REPLACE TABLE``, which BigQuery + documents drops every row access policy (and whose SELECT schema has + no policy tags). The destination is a TableReference, not SQL.""" + modules, mock_bq, mock_client = _mock_bigquery() + with modules: + result = rollback._restore_bigquery(_bq_snap(), dry_run=False) assert result["status"] == "restored" assert result["provider"] == "bigquery" + assert result["provider_result"] == {"status": "ok", "job_id": "job_restore", "rows": 3} mock_bq.Client.assert_called_once_with(project="myproj") - assert mock_client.query.call_count == 1 - executed = mock_client.query.call_args_list[0].args[0] - assert executed.startswith("CREATE OR REPLACE TABLE") - assert "AS SELECT * FROM" in executed + mock_bq.DatasetReference.assert_called_once_with("myproj", "telco") + dataset_ref = mock_bq.DatasetReference.return_value + assert mock_bq.TableReference.call_args_list[-1].args == (dataset_ref, "subscriber360") + mock_bq.QueryJobConfig.assert_called_once_with( + destination=mock_bq.TableReference.return_value, + write_disposition="WRITE_TRUNCATE_DATA", + create_disposition="CREATE_NEVER", + ) + restore = mock_client.query.call_args_list[-1] + assert restore.args == (_BQ_RESTORE_QUERY,) + assert restore.kwargs == {"job_config": mock_bq.QueryJobConfig.return_value} + assert not any( + "CREATE OR REPLACE" in c.args[0].upper() for c in mock_client.query.call_args_list + ) + + def test_a_backup_the_identity_reads_only_partly_is_refused(self): + """A row access policy on the backup (BigQuery supports row-level security + on table snapshots) shows the restoring identity some rows: ``WRITE_TRUNCATE_DATA`` + would replace the table with them. Refused before anything is written.""" + modules, _, mock_client = _mock_bigquery(rows=3, readable=2) + with modules: + with pytest.raises(CLIError, match="rollback_bigquery_backup_filtered") as raised: + rollback._restore_bigquery(_bq_snap(), dry_run=False) + assert raised.value.context["rows"] == 3 + assert raised.value.context["readable"] == 2 + executed = [c.args[0] for c in mock_client.query.call_args_list] + assert executed == ["SELECT COUNT(*) FROM `myproj`.`telco`.`subscriber360_bak_1`"] + + def test_a_backup_with_no_row_count_is_refused(self): + """Fail closed: without the backup's own count, a filtered read cannot be told + from a whole one.""" + modules, _, mock_client = _mock_bigquery(rows=None, readable=3) + with modules: + with pytest.raises(CLIError, match="rollback_bigquery_backup_filtered"): + rollback._restore_bigquery(_bq_snap(), dry_run=False) + assert len(mock_client.query.call_args_list) == 1 + + def test_on_an_emulator_the_restore_truncates_with_no_credentials(self, monkeypatch): + """goccy/bigquery-emulator appends for a disposition it does not know and + has no row access policy to keep: ``WRITE_TRUNCATE``, anonymous + credentials, and no row count (it reports none to compare).""" + from fluid_build.build_runners import _bigquery_load + + monkeypatch.setenv("BIGQUERY_EMULATOR_HOST", "http://localhost:9050") + anonymous = object() + monkeypatch.setattr(_bigquery_load, "_anonymous_credentials", lambda: anonymous) + modules, mock_bq, mock_client = _mock_bigquery() + with modules: + result = rollback._restore_bigquery(_bq_snap(), dry_run=False) + assert result["write_disposition"] == "WRITE_TRUNCATE" + assert result["provider_result"]["rows"] is None + mock_bq.Client.assert_called_once_with(project="myproj", credentials=anonymous) + assert mock_bq.QueryJobConfig.call_args.kwargs["write_disposition"] == "WRITE_TRUNCATE" + assert [c.args[0] for c in mock_client.query.call_args_list] == [_BQ_RESTORE_QUERY] def test_provider_unavailable_raises(self): """If google-cloud-bigquery isn't importable, surface a typed diff --git a/tests/providers/test_gcp_replace_keeps_governance.py b/tests/providers/test_gcp_replace_keeps_governance.py new file mode 100644 index 00000000..837a44a6 --- /dev/null +++ b/tests/providers/test_gcp_replace_keeps_governance.py @@ -0,0 +1,357 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""A replace of a governed BigQuery table must keep the table's governance. + +BigQuery documents that "when a table is replaced using the ``CREATE OR REPLACE +TABLE`` DDL statement, all existing row access policies on the original table +are dropped" (row-level security, Limitations), and the replacing table's +schema comes from the SELECT, so the column policy tags (column-level security +and data masking) go too. Measured on BigQuery, 6 October 2026: a table with a +row access policy had none after ``CREATE OR REPLACE TABLE ... AS SELECT`` and +showed every row. + +The governance (policy tags in the table schema, ``google_bigquery_row_access_policy``, +data policies) is applied by ``tofu apply`` from ``iac/providers/gcp_governance.py``, +before any build runs. A replace that recreates the table therefore leaves every +row-filtered or column-restricted principal reading everything until the next apply. +The data-only replacement that keeps them is the one the load path already uses +(``build_runners/_bigquery_load.py``): ``WRITE_TRUNCATE_DATA`` into the existing +table, which "overwrites the data, but keeps the constraints and schema of the +existing table" (BigQuery REST reference, ``JobConfigurationQuery.writeDisposition``). +""" + +from __future__ import annotations + +import copy +import logging +import re +from typing import Any, Dict, List +from unittest.mock import patch + +import pytest + +from fluid_build.providers.gcp.plan.planner import plan_actions + +pytestmark = pytest.mark.unit + +PROJECT = "fluid-probe-123" +DATASET = "silver" +TABLE = "customer_profile" + + +def _governed_contract() -> Dict[str, Any]: + """One BigQuery table with a row filter and a column restriction, and an + inline-SQL build that writes it.""" + return { + "fluidVersion": "0.7.6", + "kind": "DataProduct", + "id": "silver.customer_profile", + "name": "customer profile", + "metadata": {"owner": {"team": "data"}}, + "exposes": [ + { + "exposeId": TABLE, + "kind": "table", + "binding": { + "platform": "gcp", + "format": "bigquery_table", + "location": { + "project": PROJECT, + "dataset": DATASET, + "table": TABLE, + "region": "europe-west1", + }, + }, + "contract": { + "schema": [ + {"name": "customer_id", "type": "STRING"}, + {"name": "email", "type": "STRING"}, + {"name": "consent", "type": "BOOLEAN"}, + ] + }, + "policy": { + "authz": { + "readers": ["group:analysts@example.com"], + "rowFilters": [ + { + "principal": "group:analysts@example.com", + "name": "analysts_consented", + "where": "consent = true", + } + ], + "columnRestrictions": [ + { + "principal": "group:analysts@example.com", + "columns": ["email"], + "access": "deny", + }, + ], + } + }, + } + ], + "builds": [ + { + "id": "build_profile", + "pattern": "embedded-logic", + "engine": "sql", + "properties": { + "sql": f"SELECT customer_id, email, consent FROM `{PROJECT}.bronze.customers`" + }, + "outputs": [TABLE], + } + ], + } + + +def _actions(mode: str = "replace", contract: Dict[str, Any] | None = None) -> List[Dict]: + return plan_actions( + contract or _governed_contract(), + PROJECT, + "europe-west1", + logging.getLogger("test"), + mode=mode, + ) + + +def _replace_actions(): + return _actions("replace") + + +def _build(actions: List[Dict[str, Any]]) -> Dict[str, Any]: + (build,) = [a for a in actions if a.get("phase") == "build"] + return build + + +def _snapshot_action(actions: List[Dict[str, Any]]) -> Dict[str, Any]: + (snap,) = [a for a in actions if a.get("phase") == "snapshot"] + return snap + + +class TestReplaceModeSqlBuild: + def test_replace_does_not_recreate_the_governed_table(self): + """No planned statement may ``CREATE OR REPLACE`` the governed table: + BigQuery drops its row access policies, and the SELECT's schema has no + policy tags.""" + offending = [ + a + for a in _replace_actions() + if "CREATE OR REPLACE TABLE" in str(a.get("sql") or "").upper() + and f"{DATASET}.{TABLE}`" in str(a.get("sql")) + ] + assert offending == [], ( + "replace mode re-creates the governed table, which drops its row access " + "policies and column policy tags:\n" + "\n".join(str(a["sql"]) for a in offending) + ) + + def test_replace_keeps_the_table_definition_planned(self): + """The table (with its policy tags) stays owned by the declared table + definition in replace mode; the build replaces only its rows.""" + ensured = [ + a + for a in _replace_actions() + if a.get("op") == "bq.ensure_table" and a.get("table") == TABLE + ] + assert ensured, ( + "replace mode drops the governed table's definition from the plan and " + "leaves its materialisation to a CREATE OR REPLACE TABLE ... AS SELECT" + ) + + @pytest.mark.parametrize("mode", ["replace", "replace-and-build"]) + def test_replace_writes_the_select_into_the_existing_table(self, mode): + """The build is the SELECT, a structured destination and the query job's + dispositions: BigQuery replaces the rows and keeps the table.""" + build = _build(_actions(mode)) + assert build["op"] == "bq.sql.execute" + assert ( + build["sql"] == f"SELECT customer_id, email, consent FROM `{PROJECT}.bronze.customers`" + ) + assert build["destination"] == {"project": PROJECT, "dataset": DATASET, "table": TABLE} + assert build["write_disposition"] == "WRITE_TRUNCATE_DATA" + assert build["create_disposition"] == "CREATE_NEVER" + assert build["mode"] == "replace" + + def test_amend_still_inserts(self): + build = _build(_actions("amend")) + assert build["sql"] == ( + f"INSERT INTO `{PROJECT}.{DATASET}.{TABLE}`\n" + f"SELECT customer_id, email, consent FROM `{PROJECT}.bronze.customers`" + ) + assert "destination" not in build and "write_disposition" not in build + + @pytest.mark.parametrize( + "sql", + [ + f"INSERT INTO `{PROJECT}.{DATASET}.{TABLE}` SELECT * FROM `{PROJECT}.bronze.c`", + f"MERGE `{PROJECT}.{DATASET}.{TABLE}` t USING `{PROJECT}.bronze.c` s ON FALSE " + "WHEN NOT MATCHED THEN INSERT ROW", + " create temp table x as select 1", + ], + ids=["insert", "merge", "create"], + ) + def test_sql_that_names_its_own_sink_passes_through_untouched(self, sql): + contract = _governed_contract() + contract["builds"][0]["properties"]["sql"] = sql + build = _build(_actions("replace", contract)) + assert build["sql"] == sql + assert "destination" not in build and "write_disposition" not in build + + def test_a_crafted_destination_is_refused_not_planned(self): + """The destination is validated like every FQN the planner builds.""" + contract = _governed_contract() + contract["exposes"][0]["binding"]["location"]["dataset"] = "silver`; DROP TABLE x; --" + with pytest.raises(ValueError, match="Invalid SQL identifier"): + _actions("replace", contract) + + def test_the_backup_is_a_zero_copy_snapshot_not_an_ungoverned_copy(self): + """``CREATE TABLE … AS SELECT * FROM `` was a full copy with no row access + policy and no policy tag; a table snapshot is BigQuery's zero-copy backup.""" + snap = _snapshot_action(_replace_actions()) + backup = snap["rollback_snapshot"]["backup_name"] + assert re.fullmatch(rf"BACKUP_{TABLE}_\d+", backup) + assert snap["sql"] == ( + f"CREATE SNAPSHOT TABLE IF NOT EXISTS `{PROJECT}.{DATASET}.{backup}` " + f"CLONE `{PROJECT}.{DATASET}.{TABLE}` " + "OPTIONS(expiration_timestamp = TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 30 DAY))" + ) + assert " AS SELECT" not in snap["sql"].upper() + assert snap["allow_failure"] is True + assert snap["rollback_snapshot"]["location"] == { + "database": PROJECT, + "schema": DATASET, + "table": TABLE, + "backup_table": backup, + } + + def test_an_additive_plan_takes_no_snapshot(self): + assert not [a for a in _actions("amend") if a.get("phase") == "snapshot"] + + def test_the_plan_never_recreates_any_table(self): + for mode in ("amend", "replace", "replace-and-build"): + for action in _actions(mode): + assert "CREATE OR REPLACE" not in str(action.get("sql") or "").upper(), action + + +class TestBigQueryRollbackRestore: + """``fluid rollback`` executes the BigQuery restore; it must not recreate + the table either.""" + + def _snapshot(self) -> Dict[str, Any]: + return { + "provider": "gcp", + "backup_name": f"BACKUP_{TABLE}_1", + "location": { + "database": PROJECT, + "schema": DATASET, + "table": TABLE, + "backup_table": f"BACKUP_{TABLE}_1", + }, + } + + @staticmethod + def _client(client_cls, *, rows: int = 3, readable: int = 3): + """The patched client: a backup of ``rows`` rows, ``readable`` of them readable.""" + client = client_cls.return_value + client.get_table.return_value.num_rows = rows + + def query(sql, **_kwargs): + job = client.make_job() + job.result.return_value = [{"f0_": readable}] if "COUNT(*)" in sql else [] + return job + + client.query.side_effect = query + return client + + def test_restore_replaces_rows_not_the_table(self, monkeypatch): + bigquery = pytest.importorskip("google.cloud.bigquery") + from fluid_build.cli import rollback + + monkeypatch.delenv("BIGQUERY_EMULATOR_HOST", raising=False) + with patch.object(bigquery, "Client") as client_cls: + self._client(client_cls) + rollback._restore_bigquery(self._snapshot(), dry_run=False) + calls = client_cls.return_value.query.call_args_list + assert calls, "the restore ran no query" + for call in calls: + sql = str(call.args[0] if call.args else call.kwargs.get("query")) + assert "CREATE OR REPLACE TABLE" not in sql.upper(), ( + "the restore re-creates the governed table, dropping its row access " + f"policies and column policy tags: {sql}" + ) + # The data-only replacement: the existing table, WRITE_TRUNCATE_DATA. + assert calls[-1].args[0] == f"SELECT * FROM `{PROJECT}.{DATASET}.BACKUP_{TABLE}_1`" + job_config = calls[-1].kwargs.get("job_config") + assert job_config is not None, "the restore sends no job configuration" + assert job_config.write_disposition == "WRITE_TRUNCATE_DATA" + assert job_config.create_disposition == "CREATE_NEVER" + assert job_config.destination is not None + assert job_config.destination.project == PROJECT + assert job_config.destination.dataset_id == DATASET + assert job_config.destination.table_id == TABLE + # What BigQuery is sent: a destination table, no DDL. + assert job_config.to_api_repr()["query"]["destinationTable"] == { + "projectId": PROJECT, + "datasetId": DATASET, + "tableId": TABLE, + } + + def test_a_filtered_backup_is_not_restored(self, monkeypatch): + """The restore reads the backup as the operator: a row access policy that + shows it 2 of 3 rows would leave the table with 2. Nothing is written.""" + bigquery = pytest.importorskip("google.cloud.bigquery") + from fluid_build.cli import rollback + from fluid_build.cli._common import CLIError + + monkeypatch.delenv("BIGQUERY_EMULATOR_HOST", raising=False) + with patch.object(bigquery, "Client") as client_cls: + client = self._client(client_cls, rows=3, readable=2) + with pytest.raises(CLIError, match="rollback_bigquery_backup_filtered"): + rollback._restore_bigquery(self._snapshot(), dry_run=False) + assert all("job_config" not in c.kwargs for c in client.query.call_args_list) + + def test_rollback_restores_from_the_snapshot_the_replace_took(self, tmp_path): + """The planner's snapshot marker, recorded by the rollback writer, is what the + restore reads from: the snapshot is the query's source, the live table its + destination.""" + from fluid_build.cli import _rollback_writer, rollback + + snap_action = _snapshot_action(_replace_actions()) + with patch( + "fluid_build.providers.gcp.util.config.resolve_project_and_region", + return_value=(PROJECT, "europe-west1"), + ): + (record,) = _rollback_writer.collect_snapshots_from_actions( + [snap_action], product_id="silver.customer_profile", env="dev", provider="gcp" + ) + assert record["ddl"] == [], "no restore DDL is recorded for BigQuery" + result = rollback._restore_bigquery(copy.deepcopy(record), dry_run=True) + backup = snap_action["rollback_snapshot"]["backup_name"] + assert result["ddl"] == f"SELECT * FROM `{PROJECT}.{DATASET}.{backup}`" + assert result["destination"] == f"`{PROJECT}.{DATASET}.{TABLE}`" + assert result["create_disposition"] == "CREATE_NEVER" + + def test_provider_restore_ddl_does_not_recreate_the_table(self): + """``GcpProvider.restore_ddl`` is what ``fluid apply`` records as the + restore recipe in ``.fluid/rollback-state.json``.""" + from fluid_build.providers.gcp.provider import GcpProvider + + with patch( + "fluid_build.providers.gcp.util.config.resolve_project_and_region", + return_value=(PROJECT, "europe-west1"), + ): + provider = GcpProvider(project=PROJECT, region="europe-west1") + ddl = provider.restore_ddl(self._snapshot()) + assert not any("CREATE OR REPLACE TABLE" in s.upper() for s in ddl), ddl + assert ddl == [], "the restore is a query job, not SQL to record (as AWS)" From abd7a30fb3740146efd5c856d8255cf4853c9569 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Tue, 6 Oct 2026 20:32:53 +0200 Subject: [PATCH 3/3] docs(gcp): the replace and rollback are measured on BigQuery, not 'not proven' A probe on 6 Oct ran this branch's planner replace action and fluid rollback against a temporary governed table on real BigQuery. The replace's WRITE_TRUNCATE_DATA query job and the restore kept the row access policy, the policy tag, the partitioning and its expiration. The snapshot copied the policies and the tag. A restricted identity was refused the job (403) and nothing changed, and the old CREATE OR REPLACE dropped both on a control table. The parity doc still listed all of this as not proven. --- docs/governance-parity.md | 23 +++++++++++++++++++---- 1 file changed, 19 insertions(+), 4 deletions(-) diff --git a/docs/governance-parity.md b/docs/governance-parity.md index 50dacda5..6cd85932 100644 --- a/docs/governance-parity.md +++ b/docs/governance-parity.md @@ -382,11 +382,26 @@ data-loss gate counts, but only in a plan that already replaces the table. * `tofu plan` with terraform-provider-aws 6.67 against the live state of the lab's eleven AWS products planned no change for any of them, with only the provider pin moved from `~> 5.0`. +* **The replace and the rollback, measured on BigQuery**, 6 October 2026, on a + temporary DAY-partitioned table with a row access policy and a policy-tagged column, + driven through forge-cli's own planner action and `fluid rollback` code: + * the pre-replace `CREATE SNAPSHOT TABLE … CLONE` copied both row access policies and + the policy tag to the snapshot (not the partition expiration, as BigQuery + documents); + * the replace's `WRITE_TRUNCATE_DATA` query job, and the rollback's restore from the + snapshot, each kept the row access policy, the policy tag, the partitioning and + the partition expiration; afterwards a reader with only the filtered policy saw 2 + of 3 rows and was refused the tagged column; + * the query job maps the SELECT's columns by name, not by position, and a missing + NULLABLE column lands as NULL; + * an identity without full row access or Fine-Grained Reader is refused both the + replace job and the restore (403): it fails closed and changes nothing; + * the control, the old `CREATE OR REPLACE TABLE … AS SELECT` on an unpartitioned + copy, dropped the row access policy and the policy tag, and every row and the + tagged column became readable. On a partitioned table BigQuery refuses it outright + ("Cannot replace a table with a different partitioning spec"). * **Not proven yet**: an apply of the masking, row filter and label resources exactly - as forge-cli emits them, on either cloud. Nor, on real BigQuery, that a - `WRITE_TRUNCATE_DATA` *query* job (the rollback's and the planner's replace) keeps - the policy tags and row access policies as the measured load job does, how it maps - the SELECT's columns, or what a table snapshot keeps of its table's governance. + as forge-cli emits them, on either cloud. * **Not proven**: the Lake Formation half against a real account as 0.17.0 derives it from `columnRestrictions`. The grant shape it emits (excluded columns beside `wildcard`) was applied and enforced on a real account from 0.16.6, written by hand