fix(aws): refuse a row filter on a binding that names no Glue-catalog table

CHANGELOG.md

@@ -16,6 +16,32 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   now builds the tag's commit, also when a manual run starts from `main`, and the
   quality gate checks `ci.yml` on that commit.
 
+### Fixed
+
+- **A 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
+  `rowFilters` on a `redshift_serverless`, `redshift_table`, `redshift_external_schema`
+  or Kinesis binding, or on a Glue format with no `location.database`, passed
+  `fluid validate`, and `fluid generate iac` and `fluid apply` wrote the Redshift
+  namespace and workgroup with no filter and reported success. It is now refused
+  (`row-filter-unenforceable`), and the message names the formats that can carry a
+  filter. A mask or a deny on such a binding was already refused
+  (`column-restriction-unenforceable`), and still is.
+- **`fluid validate` reports every row-filter and mask refusal the AWS emitter makes.**
+  For an aws binding it ran only the column-restriction check, so a row filter with no
+  Lake Formation read grant or with an unmapped principal, a mask with no
+  `maskedViews`, a text mask on a column that is not a string, and a masked view over
+  a name its SQL cannot quote all passed stage 2 and failed first at
+  `fluid generate iac` or `fluid apply`. Both now run the same derivation
+  (`lf_governance`) and refuse with the same message.
+
+### Documentation
+
+- `docs/governance-parity.md` said these refusals happen at `fluid plan`. `fluid plan`
+  never runs the emitter: they happen at `fluid validate`, `fluid generate iac`,
+  `fluid apply` and the state check of `fluid diff`.
+
 ## [0.19.0] — 2026-10-06
 
 A contract can now say that a principal reads a column masked rather than not at all,

docs/governance-parity.md

@@ -17,17 +17,21 @@ contract needs `fluidVersion: "0.7.6"` to validate with them.
 | **Encryption at rest**: `binding.encryption.kms` | `product`: a KMS key per bucket (`aws_kms_key`, rotation on, alias `alias/fluid/<id>/<bucket>`) as the bucket's default SSE-KMS. `alias/...` or an ARN: that key. `none`: SSE-S3. | `product`: a key ring and key per dataset (`google_kms_key_ring` `fluid-<id>-<dataset>`, `google_kms_crypto_key` `bigquery`, 90-day rotation) in the dataset's location, `roles/cloudkms.cryptoKeyEncrypterDecrypter` for the BigQuery service agent (`google_kms_crypto_key_iam_member`), used as the dataset's `default_encryption_configuration` and the table's `encryption_configuration`. `projects/.../cryptoKeys/...`: that key. `none`: Google-managed keys. Every table of one dataset must declare the same key (see "One dataset, one key"). | AWS: the key is `Enabled`, and the objects under the prefix are SSE-KMS with it. GCP: `kmsKeyName` of the table, and of the dataset's default when the product owns the dataset. |
 | **Column restrictions**: `exposes[].policy.authz.columnRestrictions` | Each Lake Formation grant with `SELECT` excludes the restricted columns its principal may not read (`aws_lakeformation_permissions.table_with_columns.excluded_column_names`, with `wildcard`). | A Data Catalog taxonomy per product and dataset with fine-grained access control (`google_data_catalog_taxonomy`), a policy tag per set of restricted columns that share their readers (`google_data_catalog_policy_tag`), attached through the table schema's `policyTags`, and `roles/datacatalog.categoryFineGrainedReader` for exactly the allowed readers (`google_data_catalog_policy_tag_iam_member`). The restrictions' `tags` and `labels` go into the policy tag's description. | AWS: `lakeformation:ListPermissions` on tables; no `SELECT` of a denied principal, of a read grantee, or of `IAM_ALLOWED_PRINCIPALS` reaches a column it may not read, including grants made outside the contract. GCP: every restricted column carries a tag, the tag's fine-grained readers are exactly the derived set (a denied principal, or one granted outside the contract, fails), and the taxonomy enforces fine-grained access control. |
 | **Column masking**: `columnRestrictions[] {access: mask, mask: last_four \| first_four \| nullify}` | Lake Formation cannot mask. The masked principal's grant excludes the masked columns from the table, as a deny does, and a protected multi-dialect Glue Data Catalog view returns them masked (`aws_glue_catalog_table`, `table_type = VIRTUAL_VIEW`, `view_definition {is_protected, definer, representations {dialect = ATHENA, validation_connection}}`, named `<table>_masked`): exactly the columns the principal's own grant reads raw (its `columns` list, or every column less its exclusions), the masked ones through the same rule in Athena SQL, and the principal's row filter as the view's `WHERE`. The view reads the table as its definer, so it selects nothing the grant does not. The definer gets `SELECT` with grant option on the table and `CREATE_TABLE` and `DESCRIBE` on the database (Lake Formation's view prerequisites), the principal `SELECT` and `DESCRIBE` on the view. Needs `governance.lakeFormation.maskedViews {definer, validationConnection}`. | The masked columns get a policy tag of their own with a BigQuery data policy on it (`google_bigquery_datapolicy_data_policy`, `DATA_MASKING_POLICY`, predefined `LAST_FOUR_CHARACTERS`, `FIRST_FOUR_CHARACTERS` or `ALWAYS_NULL`) and `roles/bigquerydatapolicy.maskedReader` on the data policy for the masked principals (`google_bigquery_datapolicy_data_policy_iam_member`). The tag's fine-grained readers read the clear value. | AWS: the masked principal's table grant excludes the masked columns (the column check above). GCP: the masked tag's fine-grained readers are exactly the derived set. Not checked yet: the data policy and its masked readers, and the view's SQL. |
-| **Row filters**: `exposes[].policy.authz.rowFilters[] {principal, where, name?}` | A Lake Formation data cells filter per filter (`aws_lakeformation_data_cells_filter`, `row_filter.filter_expression` the predicate, `column_wildcard.excluded_column_names` the principal's excluded columns), and the principal's `SELECT` granted on the filter instead of the table. | A row access policy per filter (`google_bigquery_row_access_policy`, `filter_predicate` the predicate, `grantees` every identity of the principal, whatever it holds: a writer reads rows too), and one more, `fluid_all_rows` (`TRUE`), for every other principal the expose grants anything to, never for a filtered one, since BigQuery shows a principal no policy names no row at all and the union of those that do. | AWS: the column check reads a grant made on a data cells filter, so a filtered principal's excluded columns are still checked. Not checked yet: the filter expression on AWS, and the row access policies on GCP. |
+| **Row filters**: `exposes[].policy.authz.rowFilters[] {principal, where, name?}` | A Lake Formation data cells filter per filter (`aws_lakeformation_data_cells_filter`, `row_filter.filter_expression` the predicate, `column_wildcard.excluded_column_names` the principal's excluded columns), and the principal's `SELECT` granted on the filter instead of the table. Only on a Glue-catalog table (format `iceberg`, `parquet`, `csv`, `json`, `avro`, `orc` or `delta`, with `location.database` and `location.table`): on a Redshift or Kinesis binding the filter is refused. | A row access policy per filter (`google_bigquery_row_access_policy`, `filter_predicate` the predicate, `grantees` every identity of the principal, whatever it holds: a writer reads rows too), and one more, `fluid_all_rows` (`TRUE`), for every other principal the expose grants anything to, never for a filtered one, since BigQuery shows a principal no policy names no row at all and the union of those that do. | AWS: the column check reads a grant made on a data cells filter, so a filtered principal's excluded columns are still checked. Not checked yet: the filter expression on AWS, and the row access policies on GCP. |
 | **Governance labels**: `labels`, `exposes[].labels`, `dataClassification`, `sovereignty {jurisdiction, regulatoryFramework, dataResidency}` | The same keys as AWS tags on the bucket and the key, and as parameters on the Glue table. | GCP labels on the dataset, the table and the key: the contract's and the expose's labels, and `fluid_classification`, `fluid_jurisdiction`, `fluid_regulation` (the frameworks joined), `fluid_residency`. | Not checked yet. |
 | **Access grants**: `accessPolicy.grants[]` | Not emitted: on AWS, access is the binding's `governance.lakeFormation.grants`. `fluid validate` warns only for an aws binding with no Lake Formation grants, where the contract's access intent is unenforced. | One non-authoritative `google_bigquery_dataset_iam_member` per role and member (and `google_storage_bucket_iam_member` for GCS), with the logical principal mapped to its GCP identity. | Not checked yet on either cloud. |
 
-A policy that a binding cannot apply is refused at `fluid validate`, `fluid plan` and
-`fluid apply`, never dropped: retention, a key or a column restriction on a GCP
-binding that is not a BigQuery table (GCS, Pub/Sub, Iceberg storage), a column
-restriction on an AWS binding with no Lake Formation grants or on a non-Glue format,
-an AWS key reference on GCP and a Cloud KMS key name on AWS, a mask on an AWS binding
-with no `maskedViews` (`column-mask-unenforceable`), a text mask on a column that is
-not a string (`column-mask-type`), a row filter whose principal holds no Lake
+A policy that a binding cannot apply is refused, never dropped: at `fluid validate`
+(stage 2), and again by the emitter at `fluid generate iac`, `fluid apply` and the
+state check of `fluid diff`, before any module is written. `fluid plan` does not run
+the emitter, so it does not refuse them. Refused: retention, a key or a column
+restriction on a GCP binding that is not a BigQuery table (GCS, Pub/Sub, Iceberg
+storage), a column restriction (a deny or a mask) on an AWS binding with no Lake
+Formation grants or on a non-Glue format (`column-restriction-unenforceable`), an AWS
+key reference on GCP and a Cloud KMS key name on AWS, a mask on an AWS binding with no
+`maskedViews` (`column-mask-unenforceable`), a text mask on a column that is not a
+string (`column-mask-type`), a row filter on an AWS binding that names no Glue-catalog
+table, such as a Redshift or Kinesis format, or one whose principal holds no Lake
 Formation read grant (`row-filter-unenforceable`), and a row filter on a BigQuery view
 (`row-filter-view`) or on a GCP binding that is not a BigQuery table.
 

fluid_build/iac/governance_validation.py

@@ -14,13 +14,15 @@
 
 """Validate-time gate for the governance a cloud binding cannot apply.
 
-``fluid generate iac``, ``fluid plan`` and ``fluid apply`` refuse a principal that
-is unmapped or a placeholder, a column restriction nothing on the binding enforces,
-a Cloud KMS key on AWS or an AWS key on GCP, and retention or encryption on a GCP
-target that is not a BigQuery table. This runs the SAME derivations
-(``iac/providers/gcp_governance.py``, ``iac/column_access.py``,
-``iac/principals.py``) so ``fluid validate`` reports each refusal at stage 2, with
-the same message, instead of at apply. The same shape as ``validate_gcp_binding``.
+``fluid generate iac`` and ``fluid apply`` (and the state check of ``fluid diff``,
+which emits the same module) refuse a principal that is unmapped or a placeholder, a
+column restriction, mask or row filter nothing on the binding enforces, a Cloud KMS
+key on AWS or an AWS key on GCP, and retention or encryption on a GCP target that is
+not a BigQuery table; ``fluid plan`` does not run the emitter. This runs the SAME
+derivations (``iac/providers/gcp_governance.py``, ``lf_governance`` in
+``iac/providers/aws.py``, ``iac/column_access.py``, ``iac/principals.py``) so
+``fluid validate`` reports each refusal at stage 2, with the same message, instead of
+at apply. The same shape as ``validate_gcp_binding``.
 """
 
 from __future__ import annotations
@@ -45,6 +47,25 @@ def _lf_grants(binding: Mapping[str, Any]) -> List[Any]:
     return list(grants) if isinstance(grants, list) else []
 
 
+def _aws_refusal(exposure: Mapping[str, Any], binding: Mapping[str, Any], index: int) -> str:
+    """What the AWS emitter's ``lf_governance`` refuses this expose with, or ``""``.
+
+    A name the masked view's SQL cannot quote raises a plain ``ValueError``
+    (``validate_ident``), on which the emitter stops too. It is reported here as the
+    expose's error: escaping, it would end :func:`validate_governance`, and
+    ``fluid validate`` would report no governance finding at all.
+    """
+    from .providers.aws import lf_governance
+
+    try:
+        lf_governance(exposure, binding, index)
+    except UnsupportedBindingError as exc:
+        return _message(exc)
+    except ValueError as exc:
+        return f"exposes[{exposure.get('exposeId') or index}]: {exc}"
+    return ""
+
+
 def validate_governance(contract: Mapping[str, Any]) -> Tuple[List[str], List[str]]:
     """``(errors, warnings)`` for the contract's cloud bindings.
 
@@ -73,10 +94,10 @@ def validate_governance(contract: Mapping[str, Any]) -> Tuple[List[str], List[st
         try:
             principal_map(binding)
             if is_cloud(binding, "aws"):
-                from .providers.aws import lf_column_exclusions
-
-                lf_column_exclusions(exposure, binding, index)
-                if not _lf_grants(binding):
+                refusal = _aws_refusal(exposure, binding, index)
+                if refusal:
+                    errors.append(refusal)
+                elif not _lf_grants(binding):
                     unenforced.append(str(exposure.get("exposeId") or index))
                 continue
             if not _gov.gcp_owned(binding):

fluid_build/iac/providers/aws.py

@@ -403,14 +403,12 @@ def emit(
             fmt = binding.get("format") or "parquet"
             schema = (exposure.get("contract") or {}).get("schema") or []
             # The contract's column restrictions, as each Lake Formation grant's
-            # excluded columns (``iac/column_access.py``); refused when nothing
-            # on this binding could enforce them.
-            exclusions = lf_column_exclusions(exposure, binding, index)
-            # The contract's row filters, each on the read grant of its principal,
-            # and the protected views its masked principals read (Lake Formation
-            # cannot mask): ``lf_row_filters`` / ``lf_masked_views``.
-            row_filters = lf_row_filters(exposure, binding, index)
-            masked_views = lf_masked_views(exposure, binding, index, exclusions or {}, row_filters)
+            # excluded columns (``iac/column_access.py``), its row filters, each on
+            # the read grant of its principal, and the protected views its masked
+            # principals read (Lake Formation cannot mask); each refused when
+            # nothing on this binding could enforce it. ``fluid validate`` runs the
+            # same :func:`lf_governance`, so it refuses what this would.
+            exclusions, row_filters, masked_views = lf_governance(exposure, binding, index)
             placement = _placement(packaging, exposure)
             # The contract's governance labels (classification, jurisdiction,
             # regulation, its own labels) under fluid's own two, which win.
@@ -2096,16 +2094,29 @@ def lf_row_filters(
     A filter's principal resolves through ``binding.principals`` to its IAM identities;
     each read grant held by one of them carries the filter (:func:`_emit_lakeformation`
     grants it on a data cells filter instead of the table). A filter with no read grant
-    to carry it is refused: Lake Formation would let nobody read less.
+    to carry it is refused: Lake Formation would let nobody read less. So is a filter on
+    a binding that names no Glue-catalog table (a Redshift or Kinesis format, or no
+    ``location.database``), where :func:`_emit_lakeformation` writes nothing at all.
     """
     filters = row_filters_for(exposure, index)
     if not filters:
         return {}
     where = f"exposes[{exposure.get('exposeId') or index}].policy.authz.rowFilters"
+    loc = binding.get("location") or {}
+    fmt = str(binding.get("format") or "parquet")
+    if fmt.lower() not in _GLUE_CATALOG_FORMATS or not loc.get("database") or not loc.get("table"):
+        formats = sorted(_GLUE_CATALOG_FORMATS)
+        raise UnsupportedBindingError(
+            "row-filter-unenforceable",
+            f"{where} filters rows, but its {fmt} binding names no Glue-catalog table; the "
+            "AWS emitter enforces row filters only as Lake Formation data cells filters on "
+            f"one, a binding of format {', '.join(formats[:-1])} or {formats[-1]} with a "
+            "location.database and location.table. It would not be enforced.",
+            ("Move the filter to such an expose, or remove it from this one.",),
+        )
     gov = (binding.get("governance") or {}).get("lakeFormation") or {}
     grants = gov.get("grants") if isinstance(gov, Mapping) else None
-    loc = binding.get("location") or {}
-    if not grants or not loc.get("table"):
+    if not grants:
         raise UnsupportedBindingError(
             "row-filter-unenforceable",
             f"{where} filters rows, but this aws binding has no governance.lakeFormation "
@@ -2294,6 +2305,22 @@ def resolve(principal: str) -> Tuple[str, ...]:
     return views
 
 
+def lf_governance(
+    exposure: Mapping[str, Any], binding: Mapping[str, Any], index: int = 0
+) -> Tuple[Optional[Dict[int, Tuple[str, ...]]], Dict[int, RowFilter], List[_LfMaskedView]]:
+    """``(exclusions, row_filters, masked_views)``: one aws expose's Lake Formation policy.
+
+    :func:`lf_column_exclusions`, :func:`lf_row_filters` and :func:`lf_masked_views`, in
+    that order, as :meth:`AwsIacPlugin.emit` writes them, raising whatever it would.
+    ``fluid validate`` runs this too (``iac/governance_validation.py``), so each refusal
+    is reported at stage 2 with the emitter's own message, never first at apply.
+    """
+    exclusions = lf_column_exclusions(exposure, binding, index)
+    row_filters = lf_row_filters(exposure, binding, index)
+    masked_views = lf_masked_views(exposure, binding, index, exclusions or {}, row_filters)
+    return exclusions, row_filters, masked_views
+
+
 def _refuse_grants_left_no_column(
     exposure: Mapping[str, Any],
     binding: Mapping[str, Any],