[FEATURE](schema) Add IntegerConstraint; fail codegen on divergent union column types

packages/overture-schema-codegen/src/overture/schema/codegen/extraction/union_extraction.py

@@ -97,7 +97,7 @@ def extract_discriminator(
 
 
 _TypeShape = tuple[object, ...]
-_FieldKey = tuple[str, _TypeShape]
+_FieldKey = tuple[str, _TypeShape, frozenset[object]]
 
 
 def _structural_fingerprint(spec: FieldSpec) -> _TypeShape:
@@ -231,16 +231,20 @@ def extract_union(
         for fs in member.spec.fields:
             if fs.name in shared_field_names:
                 continue
-            key = (fs.name, _structural_fingerprint(fs))
+            # The key includes the constraints fingerprint alongside the
+            # structural one: two arms with the same name and shape but
+            # different constraints (e.g. VehicleAxleCountSelector's
+            # `ge=1 + IntegerConstraint` vs the other selectors' `ge=0`)
+            # must not collapse into one `AnnotatedField` sharing a single
+            # constraint set -- that would silently drop one arm's rules.
+            # Keeping them as separate rows, each gated to its own
+            # `variant_sources`, reuses the same per-arm `Guard` mechanism
+            # that already handles a field present on only some arms
+            # (`check_builder._field_checks_for_union`), and the renderer's
+            # collision resolver already disambiguates multiple `Check`s
+            # landing on the same field label.
+            key = (fs.name, _structural_fingerprint(fs), _constraints_fingerprint(fs))
             existing = seen.get(key)
-            if existing is not None:
-                existing_constraints = _constraints_fingerprint(existing.field_spec)
-                if _constraints_fingerprint(fs) != existing_constraints:
-                    raise ValueError(
-                        f"Union {name!r} field {fs.name!r} has the same structural "
-                        f"shape across members but diverging constraints; dedup "
-                        f"would silently drop one member's constraints"
-                    )
             prior_sources = existing.variant_sources or () if existing else ()
             seen[key] = AnnotatedField(
                 field_spec=fs,

packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/_render_common.py

@@ -33,7 +33,7 @@
 
 from overture.schema.system.field_path import ArrayPath, MapProjection
 
-from .check_ir import Check, ModelCheck
+from .check_ir import Check, Guard, ModelCheck
 from .constraint_dispatch import ForbidIf, RequireIf, model_constraint_function
 
 __all__ = [
@@ -277,13 +277,58 @@ def _symmetric_label_suffixes(keys: list[_K]) -> list[str]:
     return [f"_{idx}" if total > 1 else "" for idx, total in _occurrence_indices(keys)]
 
 
+def _field_label_suffixes(
+    keys: list[tuple[str, str, tuple[Guard, ...]]],
+) -> list[str]:
+    """Per-row field-label collision suffixes.
+
+    `keys` carries `(base_label, check_name, guards)` per emitted row. A
+    field label collides for two reasons, resolved differently:
+
+    - Across union arms -- the same field appears in several discriminator
+      arms, each carrying its own guard tuple. Every check gated to one
+      arm shares that arm's `_N` suffix (`N` its first-appearance order
+      among the label's arms), so a split field reports one consistent
+      label per arm. This includes a check unique to one arm (the axle
+      arm's `integer` check, absent from the dimension arms), which would
+      otherwise escape suffixing and report the bare label alongside its
+      `_N`-suffixed siblings.
+    - Within a single arm -- one field carries two same-named checks (a
+      lower- and upper-`bounds` pair emitted as separate checks). These
+      take a per-occurrence `_N` suffix keyed on `(label, name)`; a field
+      whose check names are all distinct stays bare.
+
+    A label reached by a single arm uses the occurrence rule (leaving
+    unsplit fields untouched); a label reached by several uses the arm
+    rule.
+    """
+    arms_by_label: dict[str, list[tuple[Guard, ...]]] = {}
+    for label, _name, guards in keys:
+        arms = arms_by_label.setdefault(label, [])
+        if guards not in arms:
+            arms.append(guards)
+    occurrences = _occurrence_indices([(label, name) for label, name, _ in keys])
+    suffixes: list[str] = []
+    for (label, _name, guards), (occ_idx, occ_total) in zip(
+        keys, occurrences, strict=True
+    ):
+        arms = arms_by_label[label]
+        if len(arms) > 1:
+            suffixes.append(f"_{arms.index(guards)}")
+        elif occ_total > 1:
+            suffixes.append(f"_{occ_idx}")
+        else:
+            suffixes.append("")
+    return suffixes
+
+
 @dataclass(frozen=True, slots=True)
 class FieldCheckRow:
     """One emitted field-check row, with its final derived strings.
 
     The renderer emits one row per descriptor of each `Check`.
     `field_check_rows` flattens the check list into these rows once,
-    computing both the symmetric `label` collision suffix and the
+    computing both the arm-grouped `label` collision suffix and the
     asymmetric `func_name` disambiguation, so the renderer and test
     renderer agree without each re-deriving them by a positional index.
 
@@ -341,15 +386,25 @@ def field_check_rows(field_checks: list[Check]) -> list[FieldCheckRow]:
             raw_func_names.append(f"_{sanitize_field_name(label)}{func_suffix}_check")
             flattened.append((check, desc_idx, label, name))
     func_names = disambiguate(raw_func_names)
-    label_suffixes = _symmetric_label_suffixes(
-        [(label, name) for _check, _idx, label, name in flattened]
+    label_suffixes = _field_label_suffixes(
+        [(label, name, check.guards) for check, _idx, label, name in flattened]
     )
-    return [
+    rows = [
         FieldCheckRow(check, desc_idx, f"{label}{label_suffix}", name, func_name)
         for (check, desc_idx, label, name), label_suffix, func_name in zip(
             flattened, label_suffixes, func_names, strict=True
         )
     ]
+    # Arm-grouped suffixing cannot distinguish two same-name checks that
+    # land in one arm of a split field; fail generation loudly if a schema
+    # ever produces that instead of emitting indistinguishable violations.
+    identities = [(row.label, row.name) for row in rows]
+    duplicates = {i for i in identities if identities.count(i) > 1}
+    if duplicates:
+        raise ValueError(
+            f"Duplicate violation identities in generated checks: {sorted(duplicates)}"
+        )
+    return rows
 
 
 @dataclass(frozen=True, slots=True)

packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/constraint_dispatch.py

@@ -18,6 +18,7 @@
 
 from overture.schema.system.case import to_snake_case
 from overture.schema.system.field_constraint.collection import UniqueItemsConstraint
+from overture.schema.system.field_constraint.numeric import IntegerConstraint
 from overture.schema.system.field_constraint.string import (
     JsonPointerConstraint,
     PatternConstraint,
@@ -350,6 +351,12 @@ def _raw_pattern(constraint: object) -> str | None:
             function="check_geometry_type", args=tuple(c.allowed_types)
         ),
     ),
+    (
+        IntegerConstraint,
+        lambda c, _bt: ExpressionDescriptor(
+            function="check_integer", constraint_type=type(c)
+        ),
+    ),
 ]
 
 

packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/schema_builder.py

@@ -16,15 +16,14 @@
     Scalar,
     UnionRef,
 )
-from ..extraction.field_walk import enum_source, terminal_scalar
+from ..extraction.field_walk import enum_source
 from ..extraction.specs import FieldSpec, ModelSpec, UnionSpec
 from ..extraction.type_registry import get_type_mapping
 
 __all__ = [
     "SHARED_TYPE_REFS",
     "SchemaField",
     "build_schema",
-    "spark_type_rank",
 ]
 
 # Types whose base_type name maps to a _schema_structs.py StructType constant.
@@ -87,58 +86,38 @@ def _spark_for_scalar(scalar: Scalar) -> str:
     return _spark_for_base(scalar.base_type, scalar.source_type)
 
 
-# Spark numeric type widening precedence (higher rank = wider type).
-_SPARK_TYPE_WIDENING: dict[str, int] = {
-    "IntegerType()": 0,
-    "LongType()": 1,
-    "DoubleType()": 2,
-}
-
-
-def spark_type_rank(field_spec: FieldSpec) -> int:
-    """Return a widening rank for the field's resolved Spark type.
-
-    Fields with a higher rank are preferred when deduplicating union
-    members by name. Non-numeric types return -1 (no widening).
-    """
-    scalar = terminal_scalar(field_spec.shape)
-    if not isinstance(scalar, Primitive):
-        return -1
-    expr = _spark_for_base(scalar.base_type, scalar.source_type)
-    return _SPARK_TYPE_WIDENING.get(expr, -1)
-
-
 def _deduplicate_by_name(fields: list[FieldSpec]) -> list[FieldSpec]:
-    """Keep one FieldSpec per name, widening the Spark type on conflict.
-
-    Union annotated_fields may contain the same field name with different
-    type shapes (e.g. `value` as uint8 in one variant and float64 in
-    another). Parquet stores one column per name, so the schema needs
-    exactly one entry. When two fields share a name, the one with the
-    wider numeric Spark type wins (matching Parquet's type-widening
-    behavior). Two same-named fields whose Spark types are non-numeric and
-    not identical cannot share a column, so the collision fails loudly
-    rather than silently keeping whichever arm came first.
+    """Keep one FieldSpec per name, requiring every arm to agree on Spark type.
+
+    Union annotated_fields may contain the same field name declared by
+    multiple arms with different `FieldSpec`s -- a `Literal` discriminator
+    whose value differs per arm, or a field whose per-arm constraints
+    diverge (see `union_extraction.extract_union`). Parquet stores one
+    column per name, so the schema needs exactly one entry. Two same-named
+    fields are compatible when they resolve to the SAME Spark type -- the
+    first-seen `FieldSpec`'s shape is kept (arbitrarily; the column type is
+    identical either way). Two same-named fields that resolve to DIFFERENT
+    Spark types cannot share a column: this always raises, whether the
+    mismatch is numeric (a narrower int type vs a float) or not, rather
+    than silently widening to the broader type -- that widening is exactly
+    the implicit Parquet-compatibility trap this check exists to surface.
     """
     seen: dict[str, FieldSpec] = {}
     for f in fields:
         existing = seen.get(f.name)
         if existing is None:
             seen[f.name] = f
             continue
-        rank_f, rank_existing = spark_type_rank(f), spark_type_rank(existing)
-        if rank_f < 0 and rank_existing < 0:
-            spark_f = _shape_to_spark(f.shape)
-            spark_existing = _shape_to_spark(existing.shape)
-            if spark_f != spark_existing:
-                raise ValueError(
-                    f"Union field {f.name!r} resolves to incompatible "
-                    f"non-widening Spark types across arms "
-                    f"({spark_existing} vs {spark_f}); a single Parquet "
-                    "column cannot represent both."
-                )
-        if rank_f > rank_existing:
-            seen[f.name] = f
+        spark_f, spark_existing = (
+            _shape_to_spark(f.shape),
+            _shape_to_spark(existing.shape),
+        )
+        if spark_f != spark_existing:
+            raise ValueError(
+                f"Union field {f.name!r} resolves to incompatible Spark "
+                f"types across arms ({spark_existing} vs {spark_f}); a "
+                "single Parquet column cannot represent both."
+            )
     return list(seen.values())
 
 

packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/test_data/base_row.py

@@ -58,7 +58,6 @@
     require_bool_field_eq,
     require_field_eq,
 )
-from ..schema_builder import spark_type_rank
 from .constraint_values import (
     CONSTRAINT_VALUES,
     curated_pattern_values,
@@ -488,27 +487,16 @@ def value_for_field(
     )
 
 
-def _widest_union_member(union: UnionSpec) -> RecordSpec:
-    """Pick the union member whose fields have the highest cumulative Spark type rank.
+def _default_union_member(union: UnionSpec) -> RecordSpec:
+    """Return the union member used when no discriminator value is known.
 
-    When multiple union members share a field name with different numeric
-    types (e.g. `value: uint8` in one variant and `value: float64` in
-    another), PySpark widens the column to the broadest type (DoubleType).
-    Generating a row from the narrower member produces Python `int` values
-    that PySpark silently converts to null in `DoubleType` columns.
-
-    By selecting the member with the widest field types, the generated row
-    uses Python `float` values that PySpark accepts in `DoubleType` columns.
+    A field shared by name across arms always resolves to the same Spark
+    type in every arm (enforced by `schema_builder._deduplicate_by_name`,
+    which raises otherwise), so any arm's synthesized value is safe to
+    write into that shared column -- the member choice is arbitrary. Picks
+    the first member, deterministically, so regeneration is stable.
     """
-    best_spec = union.member_specs[0].spec
-    best_rank = -1
-    for member in union.member_specs:
-        field_ranks = [spark_type_rank(f) for f in member.spec.fields]
-        rank = sum(r for r in field_ranks if r >= 0)
-        if rank > best_rank:
-            best_rank = rank
-            best_spec = member.spec
-    return best_spec
+    return union.member_specs[0].spec
 
 
 def resolve_arm_spec(
@@ -517,11 +505,12 @@ def resolve_arm_spec(
     """Return the member `RecordSpec` for one arm of a discriminated union.
 
     Without a discriminator value (a check not gated to a specific arm),
-    returns the widest member -- the one whose float types survive PySpark
-    column widening, per `_widest_union_member`. With a value, returns the
-    member that value selects, and raises when it selects none: a seeded
-    discriminator that matches no arm is a check_builder/scaffold inconsistency,
-    not a reason to fall back to an arm whose fields contradict the seed.
+    returns the union's first member -- safe because a field shared by name
+    across arms always resolves to the same Spark type (see
+    `_default_union_member`). With a value, returns the member that value
+    selects, and raises when it selects none: a seeded discriminator that
+    matches no arm is a check_builder/scaffold inconsistency, not a reason
+    to fall back to an arm whose fields contradict the seed.
 
     Parameters
     ----------
@@ -537,7 +526,7 @@ def resolve_arm_spec(
         When `discriminator_value` is given but selects no member arm.
     """
     if discriminator_value is None:
-        return _widest_union_member(union)
+        return _default_union_member(union)
     mapping = union.discriminator_mapping or {}
     member_cls = mapping.get(discriminator_value)  # type: ignore[call-overload]
     if member_cls is not None:
@@ -645,7 +634,7 @@ def _value_for_shape(
             # the sparse case the field is omitted from the dict and Pydantic
             # supplies the default during `TypeAdapter.validate_python()`.
             return _row_from_model_spec(
-                _widest_union_member(u),
+                _default_union_member(u),
                 index=index,
                 populate_optional=populate_optional,
             )

packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/test_data/invalid_value.py

@@ -33,6 +33,12 @@
 _INVALID_LITERALS: dict[str, object] = {
     "check_required": None,
     "check_enum": "__INVALID__",
+    # Non-integral, and >= 1 so an accompanying `ge=1` bound (the only
+    # IntegerConstraint field today, axle count) does not also fire -- the
+    # scenario isolates the integer check. A future integer field with a
+    # higher lower bound would additionally trip its bounds check, which the
+    # "at least the expected check fires" harness still tolerates.
+    "check_integer": 1.5,
     "check_url_format": "not-a-url",
     "check_url_length": "https://" + "x" * 2076,
     "check_email": "not-an-email",

packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/test_data/scaffold.py

@@ -116,7 +116,7 @@ def _child_container_spec(
     """Resolve the model a path field descends into.
 
     Returns the field's terminal `ModelRef` model, or -- for a discriminated
-    union -- the member arm the `discriminator_value` selects (the widest
+    union -- the member arm the `discriminator_value` selects (the first
     member when the check is not arm-gated). `None` when the field has neither
     a model nor a union terminal.
     """
@@ -144,7 +144,7 @@ def _walk_to_target(
     (`generate_base_row` -- required fields populated and model constraints
     such as `require_any_of` satisfied), then the on-path child overrides its
     field. A discriminated-union element resolves to the arm the seeded
-    discriminator selects (or the widest member when the check is not
+    discriminator selects (or the first member when the check is not
     arm-gated), so the element is a valid instance of a concrete arm rather
     than an untagged `{}`.
 

packages/overture-schema-codegen/tests/test_constraint_description.py

@@ -21,6 +21,7 @@
 )
 from overture.schema.codegen.extraction.specs import TypeIdentity
 from overture.schema.codegen.extraction.type_analyzer import ConstraintSource
+from overture.schema.system.field_constraint.numeric import IntegerConstraint
 from overture.schema.system.field_constraint.string import (
     CountryCodeAlpha2Constraint,
     PatternConstraint,
@@ -544,6 +545,15 @@ def link_fn(tid: TypeIdentity) -> str:
         assert received[0].obj is Target
         assert result == "References [`Target`](link) (composition)"
 
+    def test_integer_constraint_renders_docstring_and_class(self) -> None:
+        """IntegerConstraint (opaque, docstringed) flows through to prose."""
+        cs = ConstraintSource(
+            source_ref=None, source_name=None, constraint=IntegerConstraint()
+        )
+        result = constraint_display_text(cs)
+        assert "integral" in result
+        assert "`IntegerConstraint`" in result
+
 
 class TestConstraintPatternFlags:
     """constraint_display_text surfaces a compiled pattern's regex flags."""