diff --git a/packages/overture-schema-codegen/src/overture/schema/codegen/extraction/field_constraints.py b/packages/overture-schema-codegen/src/overture/schema/codegen/extraction/field_constraints.py index c62f4adbd..1cf1f2f15 100644 --- a/packages/overture-schema-codegen/src/overture/schema/codegen/extraction/field_constraints.py +++ b/packages/overture-schema-codegen/src/overture/schema/codegen/extraction/field_constraints.py @@ -11,7 +11,7 @@ import re from collections.abc import Callable -from annotated_types import Ge, Gt, Interval, Le, Lt +from annotated_types import Ge, Gt, Interval, Le, Lt, MultipleOf from overture.schema.system.primitive import GeometryTypeConstraint from overture.schema.system.ref import Reference @@ -104,6 +104,10 @@ def describe_field_constraint( result = _first_bound(constraint) if result is not None: return result + if isinstance(constraint, MultipleOf): + if constraint.multiple_of == 1: + return "Must be a whole number" + return f"Must be a multiple of {constraint.multiple_of}" if isinstance(constraint, (ArrayMinLen, ScalarMinLen)): return f"Minimum length: {constraint.min_length}" if isinstance(constraint, (ArrayMaxLen, ScalarMaxLen)): diff --git a/packages/overture-schema-codegen/src/overture/schema/codegen/extraction/union_extraction.py b/packages/overture-schema-codegen/src/overture/schema/codegen/extraction/union_extraction.py index 45c8caca8..7ab582f35 100644 --- a/packages/overture-schema-codegen/src/overture/schema/codegen/extraction/union_extraction.py +++ b/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, le=100, multiple_of=1` 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, diff --git a/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/_render_common.py b/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/_render_common.py index bfa76ca0b..0fb881a1c 100644 --- a/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/_render_common.py +++ b/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) diff --git a/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/check_builder.py b/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/check_builder.py index 56ab5499c..d89f9a586 100644 --- a/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/check_builder.py +++ b/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/check_builder.py @@ -89,6 +89,59 @@ ] +_BOUND_ORDER = ("ge", "gt", "le", "lt") + + +def _coalesce_bounds( + descriptors: list[ExpressionDescriptor], +) -> list[ExpressionDescriptor]: + """Merge multiple `check_bounds` descriptors into a single one. + + A field with both a lower and an upper bound (`Field(ge=1, le=100)`) + yields separate `Ge` and `Le` constraints, each dispatched to its own + `check_bounds`. They target the same column and describe one range, so + they collapse into one `check_bounds(ge=1, le=100)`: one violation + identity instead of two same-name checks -- which a split union arm + cannot otherwise label distinctly (see `_render_common.field_check_rows`). + + Merges only bounds of distinct kinds. Two constraints of the *same* kind + with different values (`ge=1` from a NewType, `ge=5` at the field) have no + unambiguous merge, so this raises rather than silently keeping one -- the + author should state the single intended bound. + + Raises + ------ + ValueError + When two bounds of the same kind carry different values. + """ + bound_descs = [d for d in descriptors if d.function == "check_bounds"] + if len(bound_descs) <= 1: + return descriptors + merged: dict[str, object] = {} + for d in bound_descs: + for key, value in d.kwargs: + if key in merged and merged[key] != value: + raise ValueError( + f"conflicting {key} bounds on one field: " + f"{merged[key]!r} vs {value!r}; declare a single {key}" + ) + merged[key] = value + merged_desc = ExpressionDescriptor( + function="check_bounds", + kwargs=tuple((k, merged[k]) for k in _BOUND_ORDER if k in merged), + check_nan=bound_descs[0].check_nan, + ) + result: list[ExpressionDescriptor] = [] + placed = False + for d in descriptors: + if d.function != "check_bounds": + result.append(d) + elif not placed: + result.append(merged_desc) + placed = True + return result + + def _dispatch_layer_constraints( constraints: tuple[ConstraintSource, ...], base_type: str | None, @@ -101,7 +154,7 @@ def _dispatch_layer_constraints( desc = dispatch_constraint(cs.constraint, base_type=base_type) if desc is not None: descriptors.append(desc) - return descriptors + return _coalesce_bounds(descriptors) def _literal_alternatives(shape: Scalar | MapOf) -> tuple[object, ...]: diff --git a/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/constraint_dispatch.py b/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/constraint_dispatch.py index c79c1f455..608ebd737 100644 --- a/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/constraint_dispatch.py +++ b/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/constraint_dispatch.py @@ -12,7 +12,7 @@ from dataclasses import dataclass from typing import Any, NamedTuple, TypeAlias -from annotated_types import Ge, Gt, Interval, Le, Lt +from annotated_types import Ge, Gt, Interval, Le, Lt, MultipleOf from pydantic import Strict from pydantic._internal._fields import PydanticMetadata @@ -235,6 +235,21 @@ def _dispatch_bounds( ) +def _dispatch_multiple_of( + constraint: MultipleOf, + _base_type: str | None, +) -> ExpressionDescriptor: + """Map `Field(multiple_of=n)` to a check_multiple_of descriptor. + + `check_multiple_of(col, n)` tests `col % n == 0`; `multiple_of=1` is the + integral (whole-number) case. The divisor rides in `args`, so any positive + `n` dispatches without special-casing. + """ + return ExpressionDescriptor( + function="check_multiple_of", args=(constraint.multiple_of,) + ) + + def _dispatch_pattern( constraint: PatternConstraint, _base_type: str | None, @@ -346,6 +361,7 @@ def _raw_pattern(constraint: object) -> str | None: function="check_geometry_type", args=tuple(c.allowed_types) ), ), + (MultipleOf, _dispatch_multiple_of), ] diff --git a/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/schema_builder.py b/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/schema_builder.py index 08469d67c..a4cb6b9c6 100644 --- a/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/schema_builder.py +++ b/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/schema_builder.py @@ -16,7 +16,7 @@ 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 @@ -24,7 +24,6 @@ "SHARED_TYPE_REFS", "SchemaField", "build_schema", - "spark_type_rank", ] # Types whose base_type name maps to a _schema_structs.py StructType constant. @@ -87,38 +86,28 @@ 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`). A columnar sink stores + one type per column 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 one generated column, so this + always raises, whether the mismatch is numeric (a narrower int type vs a + float) or not. + + Widening the two to their common type would often work in practice -- + Spark and Parquet can promote a narrower numeric column to a wider one + (reading an int where the schema declares a double, say). It is forbidden + anyway: a widened column makes the union's type an implicit property + inferred from whichever arms happen to disagree, rather than a decision + stated in the model. Raising forces that decision to the surface at + generation instead of leaving it as a silent compatibility trap. """ seen: dict[str, FieldSpec] = {} for f in fields: @@ -126,19 +115,16 @@ def _deduplicate_by_name(fields: list[FieldSpec]) -> list[FieldSpec]: 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()) diff --git a/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/test_data/base_row.py b/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/test_data/base_row.py index 70a20e0ee..73a263bd9 100644 --- a/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/test_data/base_row.py +++ b/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,18 @@ 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. See + `resolve_arm_spec` for why a *constraint* difference between arms never + reaches this fallback. """ - 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( @@ -516,12 +506,29 @@ def resolve_arm_spec( ) -> RecordSpec: """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. + Without a discriminator value, returns the union's first member. That + fallback is reached only for a check not gated to a specific arm, which + happens only when the check applies uniformly across arms -- so any arm + is representative and the first is a safe, deterministic choice. + + Nothing is lost by not knowing the arm here. Two arms can share a field + name at the same Spark type but with *different* constraints (axle count + is discriminated on `dimension`, and its `value` carries `ge=1, + multiple_of=1` where the other `VehicleSelector` arms carry `ge=0`). Such + divergent-constraint fields are emitted as separate arm-gated checks, so + their base rows and scaffolds always arrive WITH a discriminator and + select the correct arm below -- they never reach the first-member + default. A raise here would therefore fire on the common, correct case + (uniform shared fields), not catch a bug; the loud guards against a + divergent field slipping through un-gated live where they can see the + divergence -- `_deduplicate_by_name` (Spark-type mismatch) and the + renderer's duplicate-violation-identity check (two checks colliding on + one arm's label). + + 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 +544,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 +652,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, ) diff --git a/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/test_data/invalid_value.py b/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/test_data/invalid_value.py index e7b8c1ec9..e54081a6c 100644 --- a/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/test_data/invalid_value.py +++ b/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/test_data/invalid_value.py @@ -70,6 +70,15 @@ def invalid_value(desc: ExpressionDescriptor) -> object: return _INVALID_LITERALS[fn] if fn == "check_bounds": return invalid_bound(desc) + if fn == "check_multiple_of": + # A non-multiple of the divisor: `divisor * 1.5` leaves a remainder of + # `divisor / 2`. For divisor=1 this is 1.5, kept inside a typical + # [1, N] bound range so a co-located bounds check does not also fire + # and the scenario isolates the multiple-of check. Isolation is a + # convenience, not a requirement: the harness asserts the expected + # check is among those raised, so an extra bounds violation is + # tolerated. + return float(desc.args[0]) * 1.5 # type: ignore[arg-type] if fn == "check_pattern": if (curated := curated_pattern_values(desc)) is not None: return curated.invalid diff --git a/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/test_data/scaffold.py b/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/test_data/scaffold.py index 3d8d6554e..4d48dde05 100644 --- a/packages/overture-schema-codegen/src/overture/schema/codegen/pyspark/test_data/scaffold.py +++ b/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 `{}`. diff --git a/packages/overture-schema-codegen/tests/test_constraint_description.py b/packages/overture-schema-codegen/tests/test_constraint_description.py index 941537a2e..ff693960c 100644 --- a/packages/overture-schema-codegen/tests/test_constraint_description.py +++ b/packages/overture-schema-codegen/tests/test_constraint_description.py @@ -2,7 +2,7 @@ import re -from annotated_types import Ge, Gt, Interval, Le, Lt +from annotated_types import Ge, Gt, Interval, Le, Lt, MultipleOf from overture.schema.codegen.extraction.field_constraints import ( constraint_display_text, describe_field_constraint, @@ -544,6 +544,20 @@ def link_fn(tid: TypeIdentity) -> str: assert received[0].obj is Target assert result == "References [`Target`](link) (composition)" + def test_multiple_of_one_renders_whole_number(self) -> None: + """`Field(multiple_of=1)` renders as whole-number prose.""" + cs = ConstraintSource( + source_ref=None, source_name=None, constraint=MultipleOf(1) + ) + assert constraint_display_text(cs) == "Must be a whole number" + + def test_multiple_of_n_renders_multiple(self) -> None: + """`Field(multiple_of=n)` renders as multiple-of prose.""" + cs = ConstraintSource( + source_ref=None, source_name=None, constraint=MultipleOf(5) + ) + assert constraint_display_text(cs) == "Must be a multiple of 5" + class TestConstraintPatternFlags: """constraint_display_text surfaces a compiled pattern's regex flags.""" diff --git a/packages/overture-schema-codegen/tests/test_pyspark_base_row.py b/packages/overture-schema-codegen/tests/test_pyspark_base_row.py index a1e9ffb94..26c54da02 100644 --- a/packages/overture-schema-codegen/tests/test_pyspark_base_row.py +++ b/packages/overture-schema-codegen/tests/test_pyspark_base_row.py @@ -5,7 +5,7 @@ from typing import Any import pytest -from annotated_types import Gt, Lt +from annotated_types import Ge, Gt, Lt, MultipleOf from codegen_test_support import ( FeatureWithDict, FeatureWithRequiredUrl, @@ -27,6 +27,7 @@ ModelSpec, UnionSpec, ) +from overture.schema.codegen.extraction.union_extraction import extract_union from overture.schema.codegen.pyspark.constraint_dispatch import ExpressionDescriptor from overture.schema.codegen.pyspark.test_data.base_row import ( _primitive_default, @@ -39,8 +40,10 @@ generate_base_row, generate_populated_arm_rows, generate_populated_row, + resolve_arm_spec, value_for_field, ) +from overture.schema.common.scoping.vehicle import VehicleSelector from overture.schema.system.model_constraint import ( FieldEqCondition, forbid_if, @@ -133,6 +136,31 @@ def test_geometry_is_valid_wkt(self, connector_spec: ModelSpec) -> None: assert row["geometry"].startswith("POINT") +class TestResolveArmSpecNoDiscriminatorValue: + """Without a discriminator value, resolve_arm_spec falls back deterministically. + + Pre-widening-removal, the fallback picked the "widest" numeric member so a + narrower arm's int value wouldn't null out in a widened DoubleType column. + Now that `schema_builder._deduplicate_by_name` requires every arm to + resolve to the SAME Spark type for a shared field name, any arm's + synthesized value fits the shared column, so the fallback is just "first + member" -- exercised here via the real VehicleSelector union (five arms, + `value` now float64 on every arm). + """ + + def test_returns_first_member_deterministically(self) -> None: + spec = extract_union("VehicleAxleCountSelector", VehicleSelector) + result = resolve_arm_spec(spec) + assert result is spec.member_specs[0].spec + + def test_value_for_shape_produces_valid_row_from_fallback_arm(self) -> None: + """The fallback arm's synthesized row validates against the union.""" + spec = extract_union("VehicleAxleCountSelector", VehicleSelector) + row = generate_base_row(resolve_arm_spec(spec)) + adapter: TypeAdapter[object] = TypeAdapter(spec.source_annotation) + adapter.validate_python(row) + + class TestGenerateArmRows: def test_returns_dict_per_arm( self, segment_spec: ModelSpec, segment_union: UnionSpec @@ -587,3 +615,19 @@ def test_gt_and_lt_float_tight_interval_returns_interior_value(self) -> None: result = _value_from_scalar_constraints(scalar) assert isinstance(result, float) assert 0.0 < result < 1.0 + + def test_multiple_of_with_ge_returns_integral_value(self) -> None: + """A float64 `multiple_of=1` + ge=1 field synthesizes an integral value >= 1.""" + scalar = Primitive( + base_type="float64", + constraints=( + ConstraintSource( + source_ref=None, source_name=None, constraint=MultipleOf(1) + ), + ConstraintSource(source_ref=None, source_name=None, constraint=Ge(1)), + ), + ) + result = _value_from_scalar_constraints(scalar) + assert isinstance(result, float) + assert result >= 1 + assert result.is_integer() diff --git a/packages/overture-schema-codegen/tests/test_pyspark_check_builder.py b/packages/overture-schema-codegen/tests/test_pyspark_check_builder.py index 5fa49c72b..616e18734 100644 --- a/packages/overture-schema-codegen/tests/test_pyspark_check_builder.py +++ b/packages/overture-schema-codegen/tests/test_pyspark_check_builder.py @@ -64,6 +64,7 @@ forbid_if, require_any_of, ) +from overture.schema.system.primitive import float64 from overture.schema.system.string import CountryCodeAlpha2 from pydantic import BaseModel, Field from pydantic.fields import FieldInfo @@ -169,6 +170,37 @@ def test_optional_str_field_no_checks(self, nodes: list[Check]) -> None: assert len(name_nodes) == 0 +class _TwoSidedBoundModel(BaseModel): + v: Annotated[float64, Field(ge=1, le=100)] + + +class _ConflictingBoundModel(BaseModel): + v: Annotated[float64, Ge(1), Ge(5)] + + +class TestBoundsCoalesce: + def test_lower_and_upper_bound_merge_into_one_check_bounds(self) -> None: + # A field with both bounds yields separate Ge and Le constraints; they + # collapse into one check_bounds(ge, le) so a split union arm cannot end + # up with two indistinguishable `bounds` violation labels. + nodes, _ = _checks_for(_TwoSidedBoundModel) + bounds = [ + d + for n in nodes + if n.target == _path("v") + for d in n.descriptors + if d.function == "check_bounds" + ] + assert len(bounds) == 1 + assert dict(bounds[0].kwargs) == {"ge": 1.0, "le": 100.0} + + def test_conflicting_same_kind_bounds_raise(self) -> None: + # Two `ge` values have no unambiguous merge -- fail loud rather than + # silently keep one. + with pytest.raises(ValueError, match="conflicting ge"): + _checks_for(_ConflictingBoundModel) + + class _RequiredNewtypeModel(BaseModel): country: CountryCodeAlpha2 diff --git a/packages/overture-schema-codegen/tests/test_pyspark_constraint_dispatch.py b/packages/overture-schema-codegen/tests/test_pyspark_constraint_dispatch.py index 6b2d173e9..1e3d2275e 100644 --- a/packages/overture-schema-codegen/tests/test_pyspark_constraint_dispatch.py +++ b/packages/overture-schema-codegen/tests/test_pyspark_constraint_dispatch.py @@ -3,7 +3,7 @@ import re import pytest -from annotated_types import Ge, Gt, Interval, Le, Lt +from annotated_types import Ge, Gt, Interval, Le, Lt, MultipleOf from overture.schema.codegen.extraction.field import Primitive from overture.schema.codegen.extraction.length_constraints import ( ArrayMaxLen, @@ -313,6 +313,18 @@ def test_geometry_type(self) -> None: assert desc.function == "check_geometry_type" assert GeometryType.POINT in desc.args + def test_multiple_of_dispatches_to_check_multiple_of(self) -> None: + desc = dispatch_constraint(MultipleOf(1), base_type="float64") + assert desc is not None + assert desc.function == "check_multiple_of" + assert desc.args == (1,) + + def test_multiple_of_non_unit_divisor(self) -> None: + desc = dispatch_constraint(MultipleOf(0.5), base_type="float64") + assert desc is not None + assert desc.function == "check_multiple_of" + assert desc.args == (0.5,) + class TestSkippedConstraints: def test_reference_returns_none(self) -> None: diff --git a/packages/overture-schema-codegen/tests/test_pyspark_invalid_value.py b/packages/overture-schema-codegen/tests/test_pyspark_invalid_value.py index 9fceb373f..77a4a3ef2 100644 --- a/packages/overture-schema-codegen/tests/test_pyspark_invalid_value.py +++ b/packages/overture-schema-codegen/tests/test_pyspark_invalid_value.py @@ -53,6 +53,20 @@ def test_unknown_bound_raises(self) -> None: invalid_value(desc) +class TestInvalidValueMultipleOf: + def test_returns_non_integral_float_for_unit_divisor(self) -> None: + desc = ExpressionDescriptor(function="check_multiple_of", args=(1,)) + value = invalid_value(desc) + assert isinstance(value, float) + assert not value.is_integer() + + def test_returns_non_multiple_for_non_unit_divisor(self) -> None: + desc = ExpressionDescriptor(function="check_multiple_of", args=(0.5,)) + value = invalid_value(desc) + assert isinstance(value, float) + assert value % 0.5 != 0 + + class TestInvalidValuePattern: def test_unknown_constraint_type_raises(self) -> None: desc = ExpressionDescriptor(function="check_pattern", args=(r"^[A-Z]+$",)) diff --git a/packages/overture-schema-codegen/tests/test_pyspark_renderer.py b/packages/overture-schema-codegen/tests/test_pyspark_renderer.py index 27f0dd396..3c739932e 100644 --- a/packages/overture-schema-codegen/tests/test_pyspark_renderer.py +++ b/packages/overture-schema-codegen/tests/test_pyspark_renderer.py @@ -1125,6 +1125,44 @@ def test_labels_are_positional_not_identity_keyed(self) -> None: labels = [row.label for row in field_check_rows([first, second, first_copy])] assert labels == ["value_0", "value_1", "value_2"], labels + def test_arm_unique_check_shares_arm_suffix(self) -> None: + """A check present in only one arm takes that arm's suffix, not the bare label. + + The axle arm carries an `integer` check the dimension arms lack. + Keyed on `(field, name)` the lone `integer` row is unique, so it + would escape suffixing and report the bare `value` beside its + `value_0` siblings. Grouping by arm keeps the whole axle arm on + `value_0`. + """ + axle = (ElementGuard(discriminator="dimension", values=("axle_count",)),) + dimension = ( + ElementGuard(discriminator="dimension", values=("height", "width")), + ) + axle_required = Check( + descriptors=(ExpressionDescriptor(function="check_required"),), + target=_path("value"), + guards=axle, + ) + axle_multiple_of = Check( + descriptors=( + ExpressionDescriptor(function="check_multiple_of", args=(1,)), + ), + target=_path("value"), + guards=axle, + ) + dimension_required = Check( + descriptors=(ExpressionDescriptor(function="check_required"),), + target=_path("value"), + guards=dimension, + ) + rows = field_check_rows([axle_required, axle_multiple_of, dimension_required]) + labeled = {(row.name, row.label) for row in rows} + assert labeled == { + ("required", "value_0"), + ("multiple_of", "value_0"), + ("required", "value_1"), + }, labeled + class TestMapPathRendering: """MapPath targets render to map_keys_check / map_values_check.""" diff --git a/packages/overture-schema-codegen/tests/test_pyspark_schema_builder.py b/packages/overture-schema-codegen/tests/test_pyspark_schema_builder.py index c848d88b4..403109bd9 100644 --- a/packages/overture-schema-codegen/tests/test_pyspark_schema_builder.py +++ b/packages/overture-schema-codegen/tests/test_pyspark_schema_builder.py @@ -211,3 +211,61 @@ def test_one_schema_field_per_name(self, fields: list[SchemaField]) -> None: def test_color_field_is_string_type(self, fields: list[SchemaField]) -> None: color_field = next(f for f in fields if f.name == "color") assert color_field.type_expr == "StringType()" + + +class _NumberA(BaseModel): + pass + + +class _NumberB(BaseModel): + pass + + +class TestUnionSchemaTypeDivergence: + """build_schema raises on ANY resolved-Spark-type divergence, always -- + no numeric widening. A collision where both sides resolve to the same + Spark type (e.g. two distinct scalar types, both IntegerType) still + succeeds.""" + + def _fields_for(self, shape_a: Primitive, shape_b: Primitive) -> list[SchemaField]: + af_a = AnnotatedField( + field_spec=FieldSpec(name="value", shape=shape_a, is_required=True), + variant_sources=(_NumberA,), + ) + af_b = AnnotatedField( + field_spec=FieldSpec(name="value", shape=shape_b, is_required=True), + variant_sources=(_NumberB,), + ) + spec = UnionSpec( + name="TestNumberUnion", + description=None, + annotated_fields=[af_a, af_b], + members=[], + discriminator_field=None, + discriminator_mapping=None, + source_annotation=object(), + common_base=BaseModel, + ) + return build_schema(spec) + + def test_numeric_widening_now_raises(self) -> None: + """uint8 (IntegerType) vs float64 (DoubleType) previously widened + silently; it now raises like any other type mismatch.""" + with pytest.raises(ValueError, match="incompatible"): + self._fields_for( + Primitive(base_type="uint8"), Primitive(base_type="float64") + ) + + def test_non_numeric_mismatch_still_raises(self) -> None: + with pytest.raises(ValueError, match="incompatible"): + self._fields_for(Primitive(base_type="str"), Primitive(base_type="bool")) + + def test_same_resolved_type_does_not_raise(self) -> None: + """Two distinct scalar types resolving to the same Spark type + (e.g. int32 and uint16, both IntegerType) share a column cleanly.""" + fields = self._fields_for( + Primitive(base_type="int32"), Primitive(base_type="uint16") + ) + value_fields = [f for f in fields if f.name == "value"] + assert len(value_fields) == 1 + assert value_fields[0].type_expr == "IntegerType()" diff --git a/packages/overture-schema-codegen/tests/test_union_extraction.py b/packages/overture-schema-codegen/tests/test_union_extraction.py index 73f599332..a543c63c4 100644 --- a/packages/overture-schema-codegen/tests/test_union_extraction.py +++ b/packages/overture-schema-codegen/tests/test_union_extraction.py @@ -6,9 +6,11 @@ import pytest from annotated_types import MinLen from codegen_test_support import ( + LongNamesSegment, RailSegment, RoadSegment, SegmentBase, + ShortNamesSegment, TestEnumDiscriminatorUnion, TestSegment, TestSegmentDivergingConstraints, @@ -20,6 +22,7 @@ ConstraintSource, Primitive, ) +from overture.schema.codegen.extraction.length_constraints import ArrayMinLen from overture.schema.codegen.extraction.specs import FieldSpec, UnionSpec from overture.schema.codegen.extraction.union_extraction import ( _constraints_fingerprint, @@ -132,21 +135,50 @@ def test_discriminator_mapping_uses_enum_values(self, spec: UnionSpec) -> None: class TestDivergingConstraints: - """Same-named fields with matching shape but diverging constraints fail loudly.""" + """Same-named fields with matching shape but diverging constraints split + into separate arm-gated `AnnotatedField`s rather than raising. + + Field-level checks are already arm-gated by `Guard`s built from + `variant_sources` (see `check_builder._field_checks_for_union`), and the + renderer's collision resolver already disambiguates multiple `Check`s + that land on the same field label (e.g. the pre-existing `value_0`/ + `value_1` split for a field required only on some arms). Keeping + diverging-constraint fields as separate rows reuses that machinery + instead of dropping one arm's constraints or refusing to extract. + """ + + def test_diverging_constraints_produce_separate_annotated_fields(self) -> None: + """`ShortNamesSegment` and `LongNamesSegment` both declare `aliases` + as `list[str] | None` -- structurally identical -- but their + `min_length` constraints differ (1 vs 5). Extraction keeps them as + two `AnnotatedField`s, each gated to the arm that declared it. + """ + spec = extract_union( + "TestSegmentDivergingConstraints", TestSegmentDivergingConstraints + ) + aliases_fields = [ + af for af in spec.annotated_fields if af.field_spec.name == "aliases" + ] + assert len(aliases_fields) == 2 - def test_diverging_constraints_raise(self) -> None: - """A field shared by structure but not by constraints raises ValueError. + by_source = {af.variant_sources: af for af in aliases_fields} + assert (ShortNamesSegment,) in by_source + assert (LongNamesSegment,) in by_source - `ShortNamesSegment` and `LongNamesSegment` both declare `aliases` - as `list[str] | None`, so the structural fingerprint collapses - them — but the `min_length` constraints differ. Dedup would - silently keep one member's `FieldSpec`, so extraction raises - instead. - """ - with pytest.raises(ValueError, match="diverging constraints"): - extract_union( - "TestSegmentDivergingConstraints", TestSegmentDivergingConstraints - ) + def min_length(af: object) -> int: + for cs in af.field_spec.shape.constraints: # type: ignore[attr-defined] + if isinstance(cs.constraint, ArrayMinLen): + return cs.constraint.min_length # type: ignore[no-any-return] + raise AssertionError("no ArrayMinLen constraint found") + + assert min_length(by_source[(ShortNamesSegment,)]) == 1 + assert min_length(by_source[(LongNamesSegment,)]) == 5 + + def test_diverging_constraints_do_not_raise(self) -> None: + """Extraction succeeds where it previously raised ValueError.""" + extract_union( + "TestSegmentDivergingConstraints", TestSegmentDivergingConstraints + ) class TestUnionNameDerivation: diff --git a/packages/overture-schema-common/src/overture/schema/common/scoping/vehicle.py b/packages/overture-schema-common/src/overture/schema/common/scoping/vehicle.py index 287b25c55..82330ecb1 100644 --- a/packages/overture-schema-common/src/overture/schema/common/scoping/vehicle.py +++ b/packages/overture-schema-common/src/overture/schema/common/scoping/vehicle.py @@ -9,7 +9,7 @@ from overture.schema.common.unit import LengthUnit, WeightUnit from overture.schema.system.model_constraint import no_extra_fields -from overture.schema.system.primitive import float64, uint8 +from overture.schema.system.primitive import float64 class VehicleDimension(str, Enum): @@ -54,7 +54,17 @@ class VehicleAxleCountSelector(VehicleSelectorBase): """Selects vehicles based on the number of axles they have.""" dimension: Literal[VehicleDimension.AXLE_COUNT] - value: uint8 = Field(description="Number of axles on the vehicle") + # float64 to share the other vehicle dimensions' `value` type; `multiple_of` + # is what holds axle count to a whole number. + value: Annotated[ + float64, + Field( + ge=1, + le=100, + multiple_of=1, + description="Number of axles on the vehicle", + ), + ] @no_extra_fields diff --git a/packages/overture-schema-pyspark/src/overture/schema/pyspark/expressions/constraint_expressions.py b/packages/overture-schema-pyspark/src/overture/schema/pyspark/expressions/constraint_expressions.py index afb946d02..dd3c21fd1 100644 --- a/packages/overture-schema-pyspark/src/overture/schema/pyspark/expressions/constraint_expressions.py +++ b/packages/overture-schema-pyspark/src/overture/schema/pyspark/expressions/constraint_expressions.py @@ -116,6 +116,26 @@ def check_bounds( return F.coalesce(nan_check, *checks) +def check_multiple_of(col: Column, divisor: float) -> Column: + """Multiple-of check: a float column's value must be a multiple of `divisor`. + + Fires when `col % divisor` is non-zero. `divisor=1` is the integral + (whole-number) case -- rejecting a fractional `2.5` where a `float64` + column stands in for a count. Testing the remainder in double space + (`col % divisor != 0`) matches Pydantic's `multiple_of` exactly: for + `divisor=1`, large integral doubles like `1e30` -- whole numbers Pydantic + accepts ((1e30).is_integer() is True) -- pass, where a `floor`-based + integral check would saturate the LongType cast above 2^63 and wrongly + fire. NaN and infinity yield a NaN remainder (`!= 0`) and are rejected + directly, matching Pydantic. A null passes (presence is + `check_required`'s concern). + """ + return F.when( + col.isNotNull() & (col % divisor != 0), + error_msg(f"must be a multiple of {divisor}, got ", col.cast("string")), + ) + + def check_enum( col: Column, allowed: list[str], diff --git a/packages/overture-schema-pyspark/tests/expressions/test_constraint_expressions.py b/packages/overture-schema-pyspark/tests/expressions/test_constraint_expressions.py index 248db0cf2..b5565a4d7 100644 --- a/packages/overture-schema-pyspark/tests/expressions/test_constraint_expressions.py +++ b/packages/overture-schema-pyspark/tests/expressions/test_constraint_expressions.py @@ -17,6 +17,7 @@ check_linear_range_length, check_linear_range_order, check_min_fields_set, + check_multiple_of, check_pattern, check_radio_group, check_require_any_of, @@ -66,6 +67,69 @@ def test_except_literals_null_is_not_an_error(spark: SparkSession) -> None: assert _except_literals_error(spark, None) is None +def test_check_multiple_of_integral_float_passes(spark: SparkSession) -> None: + df = spark.createDataFrame([Row(val=2.0)], schema="val double") + result = df.select(check_multiple_of(F.col("val"), 1).alias("err")).collect() + assert result[0]["err"] is None + + +def test_check_multiple_of_negative_integral_float_passes(spark: SparkSession) -> None: + df = spark.createDataFrame([Row(val=-3.0)], schema="val double") + result = df.select(check_multiple_of(F.col("val"), 1).alias("err")).collect() + assert result[0]["err"] is None + + +def test_check_multiple_of_fractional_float_violation(spark: SparkSession) -> None: + df = spark.createDataFrame([Row(val=2.5)], schema="val double") + result = df.select(check_multiple_of(F.col("val"), 1).alias("err")).collect() + assert result[0]["err"] is not None + assert "multiple of" in result[0]["err"] + + +def test_check_multiple_of_null_passthrough(spark: SparkSession) -> None: + df = spark.createDataFrame([Row(val=None)], schema="val double") + result = df.select(check_multiple_of(F.col("val"), 1).alias("err")).collect() + assert result[0]["err"] is None + + +def test_check_multiple_of_large_integral_double_passes(spark: SparkSession) -> None: + # Integral doubles beyond 2^63 are whole numbers Pydantic accepts + # ((1e30).is_integer() is True). A floor-based check would saturate the + # LongType cast and wrongly fire; the remainder check passes them. + df = spark.createDataFrame([Row(val=1e30)], schema="val double") + result = df.select(check_multiple_of(F.col("val"), 1).alias("err")).collect() + assert result[0]["err"] is None + + +def test_check_multiple_of_nan_violation(spark: SparkSession) -> None: + df = spark.createDataFrame([Row(val=float("nan"))], schema="val double") + result = df.select(check_multiple_of(F.col("val"), 1).alias("err")).collect() + assert result[0]["err"] is not None + assert "multiple of" in result[0]["err"] + + +def test_check_multiple_of_positive_infinity_violation(spark: SparkSession) -> None: + df = spark.createDataFrame([Row(val=float("inf"))], schema="val double") + result = df.select(check_multiple_of(F.col("val"), 1).alias("err")).collect() + assert result[0]["err"] is not None + assert "multiple of" in result[0]["err"] + + +def test_check_multiple_of_negative_infinity_violation(spark: SparkSession) -> None: + df = spark.createDataFrame([Row(val=float("-inf"))], schema="val double") + result = df.select(check_multiple_of(F.col("val"), 1).alias("err")).collect() + assert result[0]["err"] is not None + assert "multiple of" in result[0]["err"] + + +def test_check_multiple_of_non_unit_divisor(spark: SparkSession) -> None: + # Divisor need not be 1: 1.5 is a multiple of 0.5, 1.75 is not. + df = spark.createDataFrame([Row(val=1.5), Row(val=1.75)], schema="val double") + result = df.select(check_multiple_of(F.col("val"), 0.5).alias("err")).collect() + assert result[0]["err"] is None + assert result[1]["err"] is not None + + def test_check_bounds_ge_le_valid(spark: SparkSession) -> None: df = spark.createDataFrame([Row(val=5)]) result = df.select(check_bounds(F.col("val"), ge=1, le=10).alias("err")).collect() diff --git a/packages/overture-schema-transportation-theme/tests/segment_baseline_schema.json b/packages/overture-schema-transportation-theme/tests/segment_baseline_schema.json index a503c7b22..1a0bfc017 100644 --- a/packages/overture-schema-transportation-theme/tests/segment_baseline_schema.json +++ b/packages/overture-schema-transportation-theme/tests/segment_baseline_schema.json @@ -1394,10 +1394,11 @@ }, "value": { "description": "Number of axles on the vehicle", - "maximum": 255, - "minimum": 0, + "maximum": 100, + "minimum": 1, + "multipleOf": 1, "title": "Value", - "type": "integer" + "type": "number" } }, "required": [