Enforce Transformation Shape Rules

Beyond schema, location, format, and row-level DML rules, a contract can constrain the shape of the transformation itself — does it aggregate by the right keys, does it join on the right columns, does it filter out what it’s supposed to. These are checked against the transformation’s translated plan, the same way everything else is — the write is aborted before anything is committed if a rule is violated.

Why structural, not a post-execution data-quality check

A plan that never groups by the declared keys, or never filters out soft-deleted rows, is a fact about the transformation’s code — true for every possible run, not something that has to be rediscovered per run by profiling the materialized output after the write already happened. Checking it here means the failure mode a rule like this guards against — “someone silently removed the GROUP BY and now the table has one row per event instead of one per customer” — is caught structurally, before any data moves, rather than downstream by a separate data-quality job that only notices after the bad table has already been written and possibly already read by something else.

The four rule types

rules:
  - type: required_group_by
    columns: [customer_id]
  - type: forbid_cross_join
  - type: required_join_columns
    columns: [order_id]
  - type: required_filter_columns
    columns: [is_deleted]

Each rule is checked against every matching node anywhere in the plan (a single write can itself contain several aggregations, joins, or filters — an aggregate feeding into a join, say) and is satisfied if any node of the relevant kind satisfies it.

required_group_by

At least one aggregation in the plan must group by every listed column — grouping by more than required (an extra partition key, say) still satisfies it. A plan that performs no aggregation at all is rejected the same as one that aggregates by the wrong keys.

forbid_cross_join

No join in the plan may be a cartesian product — an explicit CROSS JOIN, or any join with no condition at all, regardless of its declared type (Spark reports a plain, condition-less .join(other) this way too, not only .crossJoin(other)). No properties needed. This catches one of the most common and costly transformation bugs directly: a join that’s missing its ON clause silently fans every row out against every other row instead of failing loudly.

required_join_columns

At least one join’s condition must establish a genuine equality match (left.col = right.col, or the null-safe left.col <=> right.col) on every listed column — the same predicate-aware, De Morgan-/NOT-aware matching merge_condition uses for a MERGE’s ON clause, applied here to an ordinary join’s condition instead. A column merely referenced somewhere in the condition (a range check, a literal comparison, an OR branch) doesn’t satisfy it — only a required, column-to-column equality does.

required_filter_columns

At least one filter anywhere in the plan must reference each listed column — deliberately not gated on any particular predicate shape: col IS NOT NULL, col != 'deleted', col > 0, or anything else that reads the column inside a filter all satisfy it. This is weaker than required_join_columns’s equality-pairing requirement on purpose — “the plan must filter on this column somewhere” doesn’t presume what the filter should actually assert.

On Apache Beam

The Beam adapter checks these rules where a pipeline uses Beam’s schema transforms, which name the columns they use:

Rule	Satisfied by
required_filter_columns	Filter.create().whereFieldName("amount", ...) (or whereFieldNames, whereFieldId)
required_group_by	Group.byFieldNames("customer_id"), with or without .aggregateField(...); Group.globally() groups by nothing
required_join_columns	Join.innerJoin(customers).using("id") or .on(Join.FieldsEqual.left("customer_id").right("id")) (inner, outer and broadcast joins), and CoGroup.join(By.fieldNames("id"))
forbid_cross_join	violated by a CoGroup joined on no fields (By.fieldNames()), which matches every row of one side with every row of the other

A Beam SQL query (SqlTransform) is checked the same way for its GROUP BY and JOIN ... ON, because Beam builds those from the same Group and Join transforms. Its WHERE is compiled into code, so it does not satisfy required_filter_columns. Beam SQL rejects a CROSS JOIN itself when the pipeline is built.

A filter is seen under the name the column has where it filters. After RenameFields.rename("amount", "total"), a filter on total satisfies required_filter_columns: [total], not [amount]. This matches how the Spark adapter judges a filter on a renamed column.

A filter, grouping or join written as code (Filter.by, a DoFn, GroupByKey, CoGroupByKey) is opaque: Invaract cannot see which columns it uses.

• A required_* rule is not satisfied by one. A pipeline that filters only in a DoFn is rejected under required_filter_columns, the same as one that never filters.
• Under forbid_cross_join, such a step refuses the write as UNVERIFIABLE_WRITE. This covers a step that combines several collections in code (a CoGroupByKey, or a function given another collection as a side input), because any of them could be a cross join. Flatten, which only concatenates collections, is not refused.
• Schema transforms are read from Beam’s own objects. Each Beam release the adapter is tested against is checked for this. If a release changed them so the columns could not be read, a contract with any of these rules would refuse the write rather than guess.

On Beam Python pipelines

These rules can’t be checked for a Python pipeline yet. Its filters and joins are Python functions, and a beam.GroupBy’s keys are held only in pickled Python, so Beam’s description of the pipeline names no filter, join or grouping column. A contract with any of these rules is refused with UNSUPPORTED_CONTRACT_FEATURE before the pipeline runs, rather than passed unchecked.

What these rules don’t check yet

• Which specific rows a filter’s predicate actually excludes, or whether it’s trivially satisfiable (e.g. WHERE 1=1) — only that a filter referencing the declared column exists somewhere in the plan.
• A join condition expressed inside a CASE WHEN — the same limitation merge_condition has, for the same reason: that equality only holds conditionally, on some rows, not as a required part of the join.
• Which specific aggregation in a multi-stage pipeline a required_group_by rule refers to, when several aggregations exist — the rule is satisfied if any of them groups by the declared columns, not necessarily the “final” or “outermost” one.
• A custom rule type escape hatch. Unlike DML rules’ customRuleTypes, these four plan-shape types are the complete, closed set for now — there’s no plug-in mechanism yet for an organization’s own plan-shape governance vocabulary.