diff --git a/.secrets.baseline b/.secrets.baseline index b0b5aff1..2bfe2219 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": [ 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 2bf1d1d3..1a27185a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,6 +48,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 (`lakeformation-tag-definition`); and tags on a binding that names no Glue table (`lakeformation-tag-association`). The `tags` and `tagDefinitions` descriptions in fluid-schema 0.7.3 to 0.7.6 say so. +- **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`. - **A row filter on an AWS binding that is not a Glue-catalog table is refused, not dropped.** On AWS a row filter is a Lake Formation data cells filter on a Glue table, and the emitter writes no Lake Formation resource for any other format. So 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 c92534f0..b43ce6a2 100644 --- a/docs/governance-parity.md +++ b/docs/governance-parity.md @@ -188,12 +188,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 @@ -285,6 +304,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. + ## Lake Formation tags the platform owns AWS recommends one LF-tag ontology per account: a data steward or the platform @@ -337,7 +366,10 @@ to tag its own tables. A contract follows either model, per key: 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`). @@ -389,6 +421,24 @@ to tag its own tables. A contract follows either model, per key: * `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. * **Not proven**: associating a tag the platform owns on a real account, and how Lake 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/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/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/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", [])) 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)"