feat(governance): column masking, row filters and governance labels on AWS and GCP

docs/apply.md

@@ -232,7 +232,8 @@ must not land silently.
 
 When the first expose's binding is a GCP `bigquery_table`, the result is
 written as Parquet under `.fluid/staging/<build>/` and one load job moves it
-into that table (`WRITE_TRUNCATE`, `CREATE_NEVER`, the table's own schema),
+into that table (`WRITE_TRUNCATE_DATA`, which keeps the table's row access policies,
+`CREATE_NEVER`, the table's own schema),
 the load the duckdb acquisition runner performs; a `gs://` `location.path` on
 the binding is never written to. A failed or short load fails the build. Any
 other landing this path cannot write, a `gs://` or other non-S3 URI, a GCS

fluid_build/build_runners/_bigquery_load.py

@@ -40,14 +40,33 @@
 ``DATETIME`` the naive one). The wall-clock value is read as UTC, which is how
 BigQuery reads a zone-less timestamp literal too.
 
+**Row access policies.** A full refresh is ``WRITE_TRUNCATE_DATA``, not
+``WRITE_TRUNCATE``. BigQuery documents that "operations that use the
+``WRITE_TRUNCATE`` write disposition remove all existing row access policies"
+(row-level security, "Limitations"), so every build would have dropped the
+row filters ``tofu apply`` had just made, and each filtered principal would
+have read every row until the next apply. ``WRITE_TRUNCATE_DATA`` replaces
+the rows and keeps the table: its schema, policy tags and row access
+policies. It is the disposition dlt's BigQuery destination loads with for
+the same reason (``truncate-and-insert`` with ``enable_atomic_replace``).
+Measured on BigQuery, 5 October 2026: after a ``WRITE_TRUNCATE`` load the
+table had no row access policy and showed every row; after a
+``WRITE_TRUNCATE_DATA`` load it kept its policy, and on a DAY-partitioned
+table with a partition expiration it replaced the rows of every partition
+and kept the partitioning and the expiration.
+
 **Emulators.** With ``BIGQUERY_EMULATOR_HOST`` set, python-bigquery already
 sends every request to that host, but it still resolves Application Default
 Credentials first: with none it raises, and with some it sends a real OAuth
 token to a local emulator. :func:`bigquery_client` passes
 ``AnonymousCredentials`` instead, the way goccy/bigquery-emulator documents
 its clients. The goccy emulator also reports no ``outputRows`` for a load
 job, so a load whose job reports no count is checked by counting the table
-afterwards; it is never taken as zero rows loaded, nor as success.
+afterwards; it is never taken as zero rows loaded, nor as success. Its load
+handler knows ``WRITE_TRUNCATE`` and ``WRITE_EMPTY`` and appends for any
+other disposition (goccy/bigquery-emulator ``server/handler.go``), so a full
+refresh on an emulator is ``WRITE_TRUNCATE``: it has no row access policy to
+keep.
 """
 
 from __future__ import annotations
@@ -57,8 +76,9 @@
 from typing import Any, Dict, Mapping, Optional
 
 # Write disposition per source mode. The other modes (merge, dedup, CDC) need a
-# MERGE this path does not do, and are refused rather than approximated.
-_WRITE_DISPOSITION = {"full_refresh": "WRITE_TRUNCATE", "incremental_append": "WRITE_APPEND"}
+# MERGE this path does not do, and are refused rather than approximated. A full
+# refresh keeps the table's row access policies (see "Row access policies").
+_WRITE_DISPOSITION = {"full_refresh": "WRITE_TRUNCATE_DATA", "incremental_append": "WRITE_APPEND"}
 _SOURCE_FORMAT = {"parquet": "PARQUET"}
 
 # The project order ``terraform-provider-google`` reads, so the load goes to the
@@ -87,6 +107,19 @@ def emulator_host() -> Optional[str]:
     return host or None
 
 
+def write_disposition(mode: str) -> str:
+    """The load job's write disposition for ``mode``.
+
+    ``WRITE_TRUNCATE_DATA`` for a full refresh, which keeps the table's row
+    access policies, except on an emulator: goccy/bigquery-emulator appends for
+    a disposition it does not know, and it has no row access policy to keep.
+    """
+    disposition = _WRITE_DISPOSITION[mode]
+    if disposition == "WRITE_TRUNCATE_DATA" and emulator_host() is not None:
+        return "WRITE_TRUNCATE"
+    return disposition
+
+
 def bigquery_client(bigquery: Any, project: Optional[str]) -> Any:
     """A ``bigquery.Client`` for ``project``: ADC, or anonymous on an emulator.
 
@@ -272,15 +305,16 @@ def load_file(
 
     The table's own schema is passed to the job. Without it, a Parquet file
     whose columns are all optional cannot load into REQUIRED columns
-    (googleapis/python-bigquery#2373), and ``WRITE_TRUNCATE`` would replace
+    (googleapis/python-bigquery#2373), and a truncating load would replace
     the declared schema with the file's. A ``TIMESTAMP`` column the file holds
     as a naive timestamp is loaded from a UTC-adjusted copy
-    (:func:`utc_adjusted_copy`).
+    (:func:`utc_adjusted_copy`). A full refresh is ``WRITE_TRUNCATE_DATA``
+    (:func:`write_disposition`), so the table keeps its row access policies.
 
     The count is the job's ``outputRows``. A job that reports none (the goccy
-    emulator never does) is checked by counting the table: for
-    ``WRITE_TRUNCATE`` the table must then hold exactly the file's rows, and
-    for ``WRITE_APPEND`` it must have grown by them, which needs the count
+    emulator never does) is checked by counting the table: after a full
+    refresh the table must hold exactly the file's rows, and after
+    ``WRITE_APPEND`` it must have grown by them, which needs the count
     before the load; that is taken on an emulator only, and a real job with no
     count and no earlier count is refused rather than assumed.
     """
@@ -302,7 +336,7 @@ def load_file(
         before = _count_table(client, table_id, location)
     job_config = bigquery.LoadJobConfig(
         source_format=_SOURCE_FORMAT[sink_format],
-        write_disposition=_WRITE_DISPOSITION[mode],
+        write_disposition=write_disposition(mode),
         create_disposition="CREATE_NEVER",
         schema=table.schema,
     )

fluid_build/build_runners/_embedded_sql_io.py

@@ -1266,8 +1266,9 @@ def remove_staged(paths: Sequence[Path]) -> None:
 def load_bigquery_landing(landing: BigQueryLanding, *, logger: logging.Logger) -> Dict[str, Any]:
     """Load the staged result into its table: the acquisition runner's own load.
 
-    ``WRITE_TRUNCATE``: the embedded-SQL result replaces the table, as it
-    replaces the local file or the S3 object on the other targets. The count
+    ``WRITE_TRUNCATE_DATA``: the embedded-SQL result replaces the table's rows,
+    as it replaces the local file or the S3 object on the other targets, and
+    the table keeps its schema, policy tags and row access policies. The count
     the load is held to is read back from the staged file.
     """
     from ._bigquery_load import load_file
@@ -1312,7 +1313,7 @@ def write_bigquery_run_record(
     landed, as bronze is: ``records_total`` is the load's count, which
     :func:`load_bigquery_landing` has already held to the staged file's rows
     (``rows_from: write``), and the mode is ``full_refresh`` (the load is
-    ``WRITE_TRUNCATE``). It is dbt's pattern too: ``run_results.json`` keeps
+    ``WRITE_TRUNCATE_DATA``). It is dbt's pattern too: ``run_results.json`` keeps
     each node's ``relation_name`` and ``adapter_response.rows_affected``.
 
     A failed run is recorded without ``bigquery_load``: whether it changed
@@ -1496,8 +1497,8 @@ def refuse_landing_into_input(io: EmbeddedSqlIO) -> None:
     resolve to rather than on product ids: :func:`_refuse_self_consume` already
     refuses a contract consuming its own id, but an overlay naming an
     upstream's table (a typo, or two products left to the ``default``
-    dataset with the same expose id) passed it, and the build's
-    ``WRITE_TRUNCATE`` replaced another product's rows with its query result.
+    dataset with the same expose id) passed it, and the build's full-refresh
+    load replaced another product's rows with its query result.
     That is a data write, not a planned delete, so ``--allow-data-loss`` is
     never asked. An S3 landing inside the prefix an input reads is refused for
     the same reason: the object would become rows of that product's table.
@@ -1518,7 +1519,7 @@ def _refuse_loading_an_input(bq: BigQueryLanding, inputs: Sequence[ResolvedInput
                 f"{bq.table_id}, which {_entry(r.product_id, r.expose_id)} reads"
             ),
             why=(
-                f"The load replaces the table (WRITE_TRUNCATE), so {r.product_id}'s rows "
+                f"The load replaces the table's rows (WRITE_TRUNCATE_DATA), so {r.product_id}'s rows "
                 "would be overwritten with this build's query result."
             ),
             fix=(

fluid_build/build_runners/base.py

@@ -364,7 +364,7 @@ def _print_embedded_sql_io(io: Any) -> None:
         cprint(
             f"   ➡ lands BigQuery table {io.bigquery_landing.table_id} "
             f"({io.bigquery_landing.location}): staged as Parquet, then one load job "
-            "(WRITE_TRUNCATE), recorded in the build's run record",
+            "(WRITE_TRUNCATE_DATA), recorded in the build's run record",
             markup=False,
         )
     for warning in getattr(io, "warnings", None) or []:

fluid_build/cli/_verify_lf_columns.py

@@ -24,6 +24,13 @@
 them: a table-level grant (every column), a column list naming it, or a column
 wildcard that does not exclude it. So a grant made outside the contract is caught
 too. A mismatch is CRITICAL; a check that could not run is an error.
+
+A grant on a data cells filter (how a contract's ``policy.authz.rowFilters`` reach
+Lake Formation) counts too: ``ListPermissions`` with ``ResourceType: TABLE`` does not
+return those, so they are listed apart, and each filter's own columns
+(``GetDataCellsFilter``: its column list, or its column wildcard and exclusions)
+are what the grant reaches. A filter that cannot be read reaches every column, so a
+check that cannot see it fails rather than passes.
 """
 
 from __future__ import annotations
@@ -36,12 +43,65 @@
 ClientFactory = Callable[[str, str], Any]
 
 
-def _readable(resource: Mapping[str, Any], database: str, table: str) -> Optional[Set[str]]:
+def _column_spec(spec: Mapping[str, Any]) -> Set[str]:
+    """A grant's or filter's ``ColumnNames``, or its ``ColumnWildcard`` as ``{"*", "-excluded"}``.
+
+    Neither is every column (``{"*"}``).
+    """
+    if spec.get("ColumnNames"):
+        return set(spec["ColumnNames"])
+    wildcard = spec.get("ColumnWildcard") or {}
+    return {"*", *(f"-{c}" for c in (wildcard.get("ExcludedColumnNames") or ()))}
+
+
+def _filter_columns(
+    lf: Any, ref: Mapping[str, Any], cache: Dict[tuple, Optional[Set[str]]]
+) -> Set[str]:
+    """The columns a data cells filter reaches, read once per filter; every column if unreadable."""
+    key = (
+        ref.get("TableCatalogId"),
+        ref.get("DatabaseName"),
+        ref.get("TableName"),
+        ref.get("Name"),
+    )
+    if key not in cache:
+        try:
+            got = (
+                lf.get_data_cells_filter(
+                    TableCatalogId=ref.get("TableCatalogId"),
+                    DatabaseName=ref.get("DatabaseName"),
+                    TableName=ref.get("TableName"),
+                    Name=ref.get("Name"),
+                ).get("DataCellsFilter")
+                or {}
+            )
+        except Exception as exc:  # noqa: BLE001 — fail closed below
+            LOG.warning("verify_lf_columns could not read data cells filter %s: %s", key, exc)
+            got = None
+        cache[key] = None if got is None else _column_spec(got)
+    found = cache[key]
+    return {"*"} if found is None else found
+
+
+def _readable(
+    resource: Mapping[str, Any],
+    database: str,
+    table: str,
+    lf: Any = None,
+    filters: Optional[Dict[tuple, Optional[Set[str]]]] = None,
+) -> Optional[Set[str]]:
     """Columns of ``database.table`` this permission's resource reaches; ``None`` for none.
 
     The empty set with the ``"*"`` member means every column.
     """
     whole = {"*"}
+    if "DataCellsFilter" in resource:
+        ref = resource.get("DataCellsFilter") or {}
+        if ref.get("DatabaseName") != database or ref.get("TableName") != table:
+            return None
+        if lf is None:
+            return whole
+        return _filter_columns(lf, ref, filters if filters is not None else {})
     if "Table" in resource:
         ref = resource.get("Table") or {}
         if ref.get("DatabaseName") != database:
@@ -53,12 +113,7 @@ def _readable(resource: Mapping[str, Any], database: str, table: str) -> Optiona
         ref = resource.get("TableWithColumns") or {}
         if ref.get("DatabaseName") != database or ref.get("Name") != table:
             return None
-        if ref.get("ColumnNames"):
-            return set(ref["ColumnNames"])
-        wildcard = ref.get("ColumnWildcard")
-        if wildcard is not None:
-            return {"*", *(f"-{c}" for c in (wildcard.get("ExcludedColumnNames") or ()))}
-        return whole
+        return _column_spec(ref)
     return None
 
 
@@ -85,8 +140,26 @@ def _table_permissions(lf: Any) -> List[Mapping[str, Any]]:
         entries.extend(page.get("PrincipalResourcePermissions") or [])
         token = page.get("NextToken")
         if not token:
-            return entries
+            break
         kwargs["NextToken"] = token
+    # Grants on data cells filters are not TABLE permissions: an unfiltered listing
+    # returns them, as ``Resource.DataCellsFilter``.
+    kwargs = {}
+    try:
+        while True:
+            page = lf.list_permissions(**kwargs)
+            entries.extend(
+                e
+                for e in page.get("PrincipalResourcePermissions") or []
+                if "DataCellsFilter" in (e.get("Resource") or {})
+            )
+            token = page.get("NextToken")
+            if not token:
+                break
+            kwargs["NextToken"] = token
+    except Exception as exc:  # noqa: BLE001 — the TABLE listing still stands
+        LOG.warning("verify_lf_columns could not list data cells filter grants: %s", exc)
+    return entries
 
 
 def column_restrictions_dimension(
@@ -142,10 +215,11 @@ def column_restrictions_dimension(
         for g in ((binding.get("governance") or {}).get("lakeFormation") or {}).get("grants") or ()
         if isinstance(g, Mapping) and g.get("principal")
     }
+    filters: Dict[tuple, Optional[Set[str]]] = {}
     visible = {
         (entry.get("Principal") or {}).get("DataLakePrincipalIdentifier")
         for entry in permissions
-        if _readable(entry.get("Resource") or {}, database, table) is not None
+        if _readable(entry.get("Resource") or {}, database, table, lf, filters) is not None
     }
     unseen = sorted(grantees - visible)
     if unseen:
@@ -169,7 +243,7 @@ def column_restrictions_dimension(
                 continue
             if not LF_READ_PERMISSIONS & set(entry.get("Permissions") or ()):
                 continue
-            readable = _readable(entry.get("Resource") or {}, database, table)
+            readable = _readable(entry.get("Resource") or {}, database, table, lf, filters)
             if readable is None:
                 continue
             leaked.extend(c for c in _leaks(readable, list(forbidden)) if c not in leaked)