graflo.architecture.evolution.merge

Binary merge of two :class:~graflo.architecture.contract.manifest.GraphManifests.

Attributes

__all__ = ['AttachedProducer', 'DemotedKey', 'DemotedKeyPlan', 'KeyOwner', 'MergeIdentityError', 'MergeNameConflictError', 'MergeReport', 'OrderCycle', 'PinnedReference', 'SharedKeySpace', 'ambiguous_reference_refusals', 'demoted_key_refusals', 'identifier_refusals', 'key_space_refusals', 'merge_manifests', 'merge_manifests_with_report', 'origin_refusals', 'plan_demoted_key_names'] module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

AttachedProducer dataclass

A resource merge turned from creating a merged class into attaching to it.

Its steps producing the class now find it by key, write their properties onto the node found, and create none.

Source code in graflo/architecture/evolution/merge_report.py

@dataclass(frozen=True)
class AttachedProducer:
    """A resource merge turned from creating a merged class into attaching to it.

    Its steps producing the class now ``find`` it by *key*, write their
    properties onto the node found, and create none.
    """

    resource: str
    vertex: str
    side: Side
    members: tuple[str, ...]
    key: str
    """The secondary identity its records find the class by."""

Attributes

key instance-attribute

The secondary identity its records find the class by.

members instance-attribute

resource instance-attribute

side instance-attribute

vertex instance-attribute

Methods:

__init__(resource, vertex, side, members, key)

DemotedKey dataclass

A member's pre-merge key the merge demoted to a lookup-only secondary.

Source code in graflo/architecture/evolution/merge_report.py

@dataclass(frozen=True)
class DemotedKey:
    """A member's pre-merge key the merge demoted to a lookup-only secondary."""

    vertex: str
    side: Side
    member: str
    fields: tuple[str, ...]
    secondary: str
    branches: tuple[str, ...] = ()
    """Funnel branches the member's records can complete, in order, up to the
    one over its own key. Two or more mean the key no longer deduplicates."""

Attributes

branches = () class-attribute instance-attribute

Funnel branches the member's records can complete, in order, up to the one over its own key. Two or more mean the key no longer deduplicates.

fields instance-attribute

member instance-attribute

secondary instance-attribute

side instance-attribute

vertex instance-attribute

Methods:

__init__(vertex, side, member, fields, secondary, branches=())

DemotedKeyPlan dataclass

How one merged class names its members' demoted keys.

names is the secondary identity each demoted member's key answers to, by (side, member); additions the secondaries to declare for them, in order; refusals what makes the naming impossible.

Source code in graflo/architecture/evolution/merge_refusals.py

@dataclass(frozen=True)
class DemotedKeyPlan:
    """How one merged class names its members' demoted keys.

    ``names`` is the secondary identity each demoted member's key answers to,
    by ``(side, member)``; ``additions`` the secondaries to declare for them,
    in order; ``refusals`` what makes the naming impossible.
    """

    names: dict[tuple[Side, str], str]
    additions: list[SecondaryIdentity]
    refusals: list[MergeIdentityError]

Attributes

additions instance-attribute

names instance-attribute

refusals instance-attribute

Methods:

__init__(names, additions, refusals)

KeyOwner dataclass

A resource a derived identity names that produces the merged class.

It computes the class's key, so it creates and fuses its nodes.

Source code in graflo/architecture/evolution/merge_report.py

@dataclass(frozen=True)
class KeyOwner:
    """A resource a derived identity names that produces the merged class.

    It computes the class's key, so it creates and fuses its nodes.
    """

    resource: str
    vertex: str
    side: Side
    members: tuple[str, ...]

Attributes

members instance-attribute

resource instance-attribute

side instance-attribute

vertex instance-attribute

Methods:

__init__(resource, vertex, side, members)

MergeIdentityError

Bases: Refusal

A merged vertex's identity is ambiguous and nothing resolves it.

Two or more cluster members disagree on their (canonical-name) identity field-set and the VertexEquivalence declares no identity. The alternative -- silently taking the union of both field-sets as the new identity -- produces a natural key no record fully carries. Also raised for a declared identity some member cannot complete, or one whose funnel's synthetic id a member already declares as a property.

subjects names the merged class and its disagreeing members, as :func:~graflo.architecture.evolution.equivalence.subject ids; it does not appear in the message.

Source code in graflo/architecture/evolution/merge_errors.py

class MergeIdentityError(Refusal):
    """A merged vertex's identity is ambiguous and nothing resolves it.

    Two or more cluster members disagree on their (canonical-name) identity
    field-set and the ``VertexEquivalence`` declares no ``identity``. The
    alternative -- silently taking the union of both field-sets as the new
    identity -- produces a natural key no record fully carries. Also raised
    for a declared ``identity`` some member cannot complete, or one whose
    funnel's synthetic ``id`` a member already declares as a property.

    ``subjects`` names the merged class and its disagreeing members, as
    :func:`~graflo.architecture.evolution.equivalence.subject` ids; it does not
    appear in the message.
    """

    def __init__(
        self,
        message: str,
        *,
        check: RefusalKind | Literal[""] = "",
        subjects: tuple[str, ...] = (),
    ) -> None:
        # The only refusal carrying a default check: every raise site is the
        # same rule, and three of them pass no subjects either.
        super().__init__(
            message, check=check or "identity-disagreement", subjects=subjects
        )

Methods:

__init__(message, *, check='', subjects=())

Source code in graflo/architecture/evolution/merge_errors.py

def __init__(
    self,
    message: str,
    *,
    check: RefusalKind | Literal[""] = "",
    subjects: tuple[str, ...] = (),
) -> None:
    # The only refusal carrying a default check: every raise site is the
    # same rule, and three of them pass no subjects either.
    super().__init__(
        message, check=check or "identity-disagreement", subjects=subjects
    )

MergeNameConflictError

Bases: Refusal

Two names denote one concept under different naming conventions.

Distinct from MergeCanonicalConflictError in canonical.py, which reports a declared CanonicalMap contradicting the op. This one fires on the residue neither side declared -- the undeclared path, where merge would otherwise produce two unrelated types with the data split between them and nothing raising.

check names the rule that refused and subjects the names it is about, as :func:~graflo.architecture.evolution.equivalence.subject ids; see :class:.Refusal.

Source code in graflo/architecture/evolution/merge_errors.py

class MergeNameConflictError(Refusal):
    """Two names denote one concept under different naming conventions.

    Distinct from ``MergeCanonicalConflictError`` in ``canonical.py``, which
    reports a *declared* CanonicalMap contradicting the op. This one fires on
    the residue neither side declared -- the undeclared path, where merge
    would otherwise produce two unrelated types with the data split between
    them and nothing raising.

    ``check`` names the rule that refused and ``subjects`` the names it is
    about, as :func:`~graflo.architecture.evolution.equivalence.subject` ids;
    see :class:`.Refusal`.
    """

MergeReport dataclass

What a merge did to the sources, beyond the manifest it returns.

Source code in graflo/architecture/evolution/merge_report.py

@dataclass(frozen=True)
class MergeReport:
    """What a merge did to the sources, beyond the manifest it returns."""

    demoted: list[DemotedKey] = field(default_factory=list)
    owners: list[KeyOwner] = field(default_factory=list)
    attached: list[AttachedProducer] = field(default_factory=list)
    references: list[PinnedReference] = field(default_factory=list)
    shared_key_spaces: list[SharedKeySpace] = field(default_factory=list)
    order_cycles: list[OrderCycle] = field(default_factory=list)

Attributes

attached = field(default_factory=list) class-attribute instance-attribute

demoted = field(default_factory=list) class-attribute instance-attribute

order_cycles = field(default_factory=list) class-attribute instance-attribute

owners = field(default_factory=list) class-attribute instance-attribute

references = field(default_factory=list) class-attribute instance-attribute

shared_key_spaces = field(default_factory=list) class-attribute instance-attribute

Methods:

__init__(demoted=list(), owners=list(), attached=list(), references=list(), shared_key_spaces=list(), order_cycles=list())

OrderCycle dataclass

A resource order constraint the union's resource order leaves unmet.

resource depends on before for vertex -- it attaches to or references what before writes -- but runs first, because the constraints formed a cycle and the declared order was kept.

Source code in graflo/architecture/evolution/merge_report.py

@dataclass(frozen=True)
class OrderCycle:
    """A resource order constraint the union's resource order leaves unmet.

    *resource* depends on *before* for *vertex* -- it attaches to or
    references what *before* writes -- but runs first, because the
    constraints formed a cycle and the declared order was kept.
    """

    resource: str
    before: str
    vertex: str

Attributes

before instance-attribute

resource instance-attribute

vertex instance-attribute

Methods:

__init__(resource, before, vertex)

PinnedReference dataclass

A resource that only references a re-keyed member, pointed at its key.

Its edge endpoints for the class now select key.

Source code in graflo/architecture/evolution/merge_report.py

@dataclass(frozen=True)
class PinnedReference:
    """A resource that only references a re-keyed member, pointed at its key.

    Its edge endpoints for the class now select *key*.
    """

    resource: str
    vertex: str
    side: Side
    members: tuple[str, ...]
    key: str

Attributes

key instance-attribute

members instance-attribute

resource instance-attribute

side instance-attribute

vertex instance-attribute

Methods:

__init__(resource, vertex, side, members, key)

SharedKeySpace dataclass

Members of one side whose demoted keys are one secondary identity.

Their key values are taken to be one space: an id of one member finds a node of another.

Source code in graflo/architecture/evolution/merge_report.py

@dataclass(frozen=True)
class SharedKeySpace:
    """Members of one side whose demoted keys are one secondary identity.

    Their key values are taken to be one space: an id of one member finds a
    node of another.
    """

    vertex: str
    side: Side
    members: tuple[str, ...]
    key: str

Attributes

key instance-attribute

members instance-attribute

side instance-attribute

vertex instance-attribute

Methods:

__init__(vertex, side, members, key)

Functions:

ambiguous_reference_refusals(resolution, sides, member_properties)

Each resource that would have to find one class by two demoted keys.

The rule :func:_attach_uncovered_producers and :func:_pin_member_references refuse on, read from the sides before the union: a resource producing members of one side whose keys are demoted to different secondaries -- two key spaces, or two field sets -- when it only references them (every step lookup_only), or when it would be attached: it writes them, the derived identity names it nowhere, and its records complete no property branch. sides carry the union's resource names; member_properties are each member's properties in canonical names.

Source code in graflo/architecture/evolution/merge_refusals.py

def ambiguous_reference_refusals(
    resolution: ClusterResolution,
    sides: Mapping[str, GraphManifest],
    member_properties: Mapping[tuple[Side, str], set[str]],
) -> list[MergeIdentityError]:
    """Each resource that would have to find one class by two demoted keys.

    The rule :func:`_attach_uncovered_producers` and
    :func:`_pin_member_references` refuse on, read from the sides before the
    union: a resource producing members of one side whose keys are demoted to
    different secondaries -- two key spaces, or two field sets -- when it
    only references them (every step ``lookup_only``), or when it would be
    attached: it writes them, the derived identity names it nowhere, and its
    records complete no property branch. *sides* carry the union's resource
    names; *member_properties* are each member's properties in canonical
    names.
    """
    names = {
        member: name
        for plan in _demoted_key_plans(
            resolution, {"left": sides["left"], "right": sides["right"]}
        )
        for member, name in plan.names.items()
    }
    out: list[MergeIdentityError] = []
    for cluster in resolution.index.vertices:
        declaration = cluster.declaration
        if (
            not isinstance(declaration, VertexEquivalence)
            or declaration.identity is None
            or declaration.retire != "demote"
        ):
            continue
        local_key = declaration.local_key_branch()
        owners = {
            *declaration.derive_at,
            *(r for branch in declaration.derivations() for r in branch.sources),
            *(local_key.local_key if local_key is not None else ()),
        }
        raw = declaration.raw_branches()
        for side in ("left", "right"):
            manifest = sides[side]
            ingestion, schema = manifest.ingestion_model, manifest.graph_schema
            if ingestion is None or schema is None:
                continue
            known = schema.core_schema.vertex_config.vertex_set
            for resource in ingestion.resources:
                produced = {
                    member: steps
                    for member in cluster.members(side)
                    if (
                        steps := _steps_producing(
                            resource.pipeline, member, known_vertices=known
                        )
                    )
                }
                keys = {names.get((side, member)) for member in produced}
                named = sorted(key for key in keys if key is not None)
                if len(named) < 2 or any(
                    step_finds(step, member, known_vertices=known) is not None
                    for member, steps in produced.items()
                    for step in steps
                ):
                    continue
                references = all(
                    step_looks_up(step, member)
                    for member, steps in produced.items()
                    for step in steps
                )
                attached = (
                    not references
                    and declaration.has_derivation
                    and resource.name not in owners
                    and None not in keys
                    and not (
                        raw
                        and all(
                            any(
                                set(fields)
                                <= member_properties.get((side, member), set())
                                for fields in raw
                            )
                            for member in produced
                        )
                    )
                )
                if not references and not attached:
                    continue
                members = sorted(produced)
                out.append(
                    MergeIdentityError(
                        f"merge_manifests: resource {resource.name!r} "
                        f"{'references' if references else 'upserts'} {side} "
                        f"members {members} of {cluster.into!r}, whose keys are "
                        f"demoted to different secondary identities {named}; it "
                        "cannot find the class by one of them. Tag the members' "
                        "`local_key` alike (one key space), or produce each "
                        "member from its own resource.",
                        check="ambiguous-reference",
                        subjects=(
                            subject("merged", cluster.into),
                            *(subject(side, member) for member in members),
                        ),
                    )
                )
    return out

demoted_key_refusals(resolution, manifests)

Every demoted key :func:plan_demoted_key_names cannot name.

Read from the sides before the union, so the preview and the merge refuse alike before anything is renamed.

Source code in graflo/architecture/evolution/merge_refusals.py

def demoted_key_refusals(
    resolution: ClusterResolution, manifests: Mapping[Side, GraphManifest]
) -> list[MergeIdentityError]:
    """Every demoted key :func:`plan_demoted_key_names` cannot name.

    Read from the sides before the union, so the preview and the merge refuse
    alike before anything is renamed.
    """
    return [
        refusal
        for plan in _demoted_key_plans(resolution, manifests)
        for refusal in plan.refusals
    ]

identifier_refusals(resolution, manifests)

Every demoted-key property name a PostgreSQL target would truncate.

A demoted key field is renamed <key space>__<field>; PostgreSQL keeps only the first 63 bytes of a column name, so two such names could become one column. Checked only when the union targets PostgreSQL. Read from the sides before the union, so the preview and the merge refuse alike.

Source code in graflo/architecture/evolution/merge_refusals.py

def identifier_refusals(
    resolution: ClusterResolution, manifests: Mapping[Side, GraphManifest]
) -> list[MergeIdentityError]:
    """Every demoted-key property name a PostgreSQL target would truncate.

    A demoted key field is renamed ``<key space>__<field>``; PostgreSQL keeps
    only the first 63 bytes of a column name, so two such names could
    become one column. Checked only when the union targets PostgreSQL. Read
    from the sides before the union, so the preview and the merge refuse
    alike.
    """
    if _union_db_flavor(manifests) != DBType.POSTGRES:
        return []
    into_of = {
        (side, member): cluster.into
        for cluster in resolution.index.vertices
        for side in ("left", "right")
        for member in cluster.members(side)
    }
    out: list[MergeIdentityError] = []
    for (side, member), space in sorted(resolution.key_spaces.items()):
        renames = resolution.side_maps[side].properties.get(member, {})
        for own, name in renames.items():
            size = len(name.encode("utf-8"))
            if name != f"{space}__{own}" or size <= _PG_IDENTIFIER_BYTES:
                continue
            into = into_of.get((side, member), member)
            tag = (
                f"`origins.{side}`"
                if space == resolution.origins[side]
                else "the member's `local_key` tag"
            )
            out.append(
                MergeIdentityError(
                    f"merge_manifests: identifier: the key {side}:{member}.{own} "
                    f"demotes on {into!r} would be the property {name!r}, "
                    f"{size} bytes, but PostgreSQL keeps only "
                    f"{_PG_IDENTIFIER_BYTES}; set a shorter {tag}",
                    check="identifier",
                    subjects=(subject("merged", into),),
                )
            )
    return out

key_space_refusals(resolution, manifests)

Every problem with the key spaces resolution names demoted keys by.

A local_key tag names the key space of a member's own ids, and its demoted key is named by it. Refused: a member two of its resources tag differently (its ids would be in two spaces at once), a tag that cannot name a key space (see :func:~graflo.architecture.evolution.naming_graph.key_space_name_problem), a tag a member of the cluster already declares as a secondary identity (the demoted key would take that name), and a key space both sides name (two sides' ids in one space). An origin colliding with a secondary is :func:origin_refusals' to report.

Source code in graflo/architecture/evolution/merge_refusals.py

def key_space_refusals(
    resolution: ClusterResolution, manifests: Mapping[Side, GraphManifest]
) -> list[MergeIdentityError]:
    """Every problem with the key spaces *resolution* names demoted keys by.

    A ``local_key`` tag names the key space of a member's own ids, and its
    demoted key is named by it. Refused: a member two of its resources tag
    differently (its ids would be in two spaces at once), a tag that cannot
    name a key space (see :func:`~graflo.architecture.evolution.naming_graph.key_space_name_problem`),
    a tag a member of the cluster already declares as a secondary identity
    (the demoted key would take that name), and a key space both sides name
    (two sides' ids in one space). An origin colliding with a secondary is
    :func:`origin_refusals`' to report.
    """
    origins = resolution.origins
    clusters = {
        (side, member): cluster.into
        for cluster in resolution.index.vertices
        for side in ("left", "right")
        for member in cluster.members(side)
    }
    out: list[MergeIdentityError] = []
    authored: dict[str, dict[str, list[str]]] = {}
    for cluster in resolution.index.vertices:
        for member_side in ("left", "right"):
            schema = manifests[member_side].graph_schema
            if schema is None:
                continue
            vertex_config = schema.core_schema.vertex_config
            for member in cluster.members(member_side):
                if member not in vertex_config.vertex_set:
                    continue
                for entry in vertex_config[member].secondary_identities:
                    authored.setdefault(cluster.into, {}).setdefault(
                        entry.name, []
                    ).append(f"{member_side}:{member}")
    for (side, member), space in sorted(resolution.key_spaces.items()):
        into = clusters.get((side, member))
        authors = authored.get(into or "", {}).get(space)
        if space == origins[side] or not authors:
            continue
        out.append(
            MergeIdentityError(
                f"merge_manifests: key space: {side}:{member} is tagged "
                f"{space!r}, which names its demoted key on {into!r}, but "
                f"{', '.join(sorted(authors))} declare a secondary identity "
                f"named {space!r}; give the member another tag, or rename that "
                "secondary",
                check="key-space",
                subjects=(subject("merged", into or member), subject(side, member)),
            )
        )
    for (side, member), tags in sorted(resolution.key_tags.items()):
        nodes = (
            subject("merged", clusters.get((side, member), member)),
            subject(side, member),
        )
        if len(tags) > 1:
            out.append(
                MergeIdentityError(
                    f"merge_manifests: key space: {side}:{member} is tagged "
                    f"{list(tags)} by its resources' local_key entries, so its "
                    "ids would be in several key spaces; give the member one tag",
                    check="key-space",
                    subjects=nodes,
                )
            )
        elif tags[0] != origins[side] and (problem := key_space_name_problem(tags[0])):
            out.append(
                MergeIdentityError(
                    f"merge_manifests: key space: {side}:{member} is tagged "
                    f"{tags[0]!r}, which names its demoted key but {problem}; "
                    "give the member another tag",
                    check="key-space",
                    subjects=nodes,
                )
            )
    named: dict[str, set[Side]] = {}
    for (side, _member), space in resolution.key_spaces.items():
        named.setdefault(space, set()).add(side)
    for space, sides in sorted(named.items()):
        # Two equal origins are the origin rule's refusal.
        if len(sides) < 2 or space == origins["left"] == origins["right"]:
            continue
        members = sorted(
            (side, member)
            for (side, member), name in resolution.key_spaces.items()
            if name == space
        )
        named_members = ", ".join(f"{side}:{member}" for side, member in members)
        out.append(
            MergeIdentityError(
                f"merge_manifests: key space: {named_members} would share "
                f"the key space {space!r}, though they come from different "
                "sides; tag one side's members otherwise, or set `origins`",
                check="key-space",
                subjects=tuple(subject(side, member) for side, member in members),
            )
        )
    return out

merge_manifests(left, right, op, *, bump_version='minor', finish_init=True, strict_references=False, dynamic_edge_feedback=False, canonical_maps=())

Return a new manifest that is the deterministic merge of left and right.

The declared clusters (each a :class:~graflo.architecture.evolution.ops.VertexEquivalence or :class:~graflo.architecture.evolution.ops.RelationEquivalence, possibly n-ary) and op.canonical_maps are resolved together into one composite :class:~graflo.architecture.evolution.ops.CanonicalizeOp per side (see :func:~graflo.architecture.evolution.canonical.resolve_clusters), applied to that side in one step, before the two sides are unioned by name. Does not invent semantic matches: a name both sides carry and no cluster merges is refused under name_conflict="error" (naming the equivalences to declare), synthesized into a 1-1 cluster under union_right so it reconciles exactly as a declared one, and kept apart under prefix_right.

A vertex equivalence whose identity has a derived or local_key branch further rewrites the merged union with the fundamental ops that identity lowers to (see :func:~graflo.architecture.evolution.alignment.identity_to_ops); member-keyed sources are resolved against the sides as handed in. canonical_maps — (side, CanonicalMap) pairs — are folded into op.canonical_maps; putting the maps on the op itself keeps the whole recipe in one document.

Source code in graflo/architecture/evolution/merge.py

def merge_manifests(
    left: GraphManifest,
    right: GraphManifest,
    op: MergeManifestsOp,
    *,
    bump_version: bool | Literal["minor"] = "minor",
    finish_init: bool = True,
    strict_references: bool = False,
    dynamic_edge_feedback: bool = False,
    canonical_maps: Sequence[tuple[Side, CanonicalMap]] = (),
) -> GraphManifest:
    """Return a new manifest that is the deterministic merge of *left* and *right*.

    The declared clusters (each a :class:`~graflo.architecture.evolution.ops.VertexEquivalence`
    or :class:`~graflo.architecture.evolution.ops.RelationEquivalence`, possibly
    n-ary) and ``op.canonical_maps`` are resolved together into one composite
    :class:`~graflo.architecture.evolution.ops.CanonicalizeOp` per side
    (see :func:`~graflo.architecture.evolution.canonical.resolve_clusters`),
    applied to that side in one step, before the two sides are unioned by
    name. Does not invent semantic matches: a name both sides carry and no
    cluster merges is refused under ``name_conflict="error"`` (naming the
    equivalences to declare), synthesized into a 1-1 cluster under
    ``union_right`` so it reconciles exactly as a declared one, and kept apart
    under ``prefix_right``.

    A vertex equivalence whose ``identity`` has a derived or ``local_key``
    branch further rewrites the merged union with the fundamental ops that
    identity lowers to (see
    :func:`~graflo.architecture.evolution.alignment.identity_to_ops`);
    member-keyed sources are resolved against the sides as handed in.
    *canonical_maps* — ``(side, CanonicalMap)`` pairs — are folded into
    ``op.canonical_maps``; putting the maps on the op itself keeps the whole
    recipe in one document.
    """
    manifest, _report = _merge_manifests(
        left,
        right,
        op,
        bump_version=bump_version,
        finish_init=finish_init,
        strict_references=strict_references,
        dynamic_edge_feedback=dynamic_edge_feedback,
        canonical_maps=canonical_maps,
    )
    return manifest

merge_manifests_with_report(left, right, op, *, bump_version='minor', finish_init=True, strict_references=False, dynamic_edge_feedback=False, canonical_maps=())

:func:merge_manifests, and the :class:MergeReport of what it did.

The report lists the key owners, the resources attached by their own key, the references pinned to a demoted key, the shared key spaces, the demoted keys and any resource order constraint a cycle left unmet.

Source code in graflo/architecture/evolution/merge.py

def merge_manifests_with_report(
    left: GraphManifest,
    right: GraphManifest,
    op: MergeManifestsOp,
    *,
    bump_version: bool | Literal["minor"] = "minor",
    finish_init: bool = True,
    strict_references: bool = False,
    dynamic_edge_feedback: bool = False,
    canonical_maps: Sequence[tuple[Side, CanonicalMap]] = (),
) -> tuple[GraphManifest, MergeReport]:
    """:func:`merge_manifests`, and the :class:`MergeReport` of what it did.

    The report lists the key owners, the resources attached by their own key,
    the references pinned to a demoted key, the shared key spaces, the
    demoted keys and any resource order constraint a cycle left unmet.
    """
    return _merge_manifests(
        left,
        right,
        op,
        bump_version=bump_version,
        finish_init=finish_init,
        strict_references=strict_references,
        dynamic_edge_feedback=dynamic_edge_feedback,
        canonical_maps=canonical_maps,
    )

origin_refusals(op, manifests, resolution)

Every problem with the origins resolution names something by.

Only a side whose origin names something is checked: a demoted key whose key space is the origin, or a local key the origin tags (see :attr:~graflo.architecture.evolution.canonical.ClusterResolution.origin_tagged). Its origin must be a valid key space name (see :func:~graflo.architecture.evolution.naming_graph.key_space_name_problem; it is joined to field names with __); two named origins must differ; and no member of a re-keyed cluster may declare a secondary identity of that name, which the demoted key takes. Each refusal is about the merged classes that name something by the origin, or that declare the secondary.

Source code in graflo/architecture/evolution/merge_refusals.py

def origin_refusals(
    op: MergeManifestsOp,
    manifests: Mapping[Side, GraphManifest],
    resolution: ClusterResolution,
) -> list[MergeIdentityError]:
    """Every problem with the origins *resolution* names something by.

    Only a side whose origin names something is checked: a demoted key whose
    key space is the origin, or a local key the origin tags (see
    :attr:`~graflo.architecture.evolution.canonical.ClusterResolution.origin_tagged`).
    Its origin must be a valid key space name (see
    :func:`~graflo.architecture.evolution.naming_graph.key_space_name_problem`;
    it is joined to field names with ``__``); two named origins must differ;
    and no member of a re-keyed cluster may declare a secondary identity of
    that name, which the demoted key takes. Each refusal is about the merged
    classes that name something by the origin, or that declare the secondary.
    """
    sides: tuple[Side, ...] = ("left", "right")
    origins = resolution.origins
    into_of = {
        (side, member): cluster.into
        for cluster in resolution.index.vertices
        for side in sides
        for member in cluster.members(side)
    }
    demoting: dict[Side, set[str]] = {side: set() for side in sides}
    for (side, member), space in resolution.key_spaces.items():
        into = into_of.get((side, member))
        if into is not None and space == origins[side]:
            demoting[side].add(into)
    tagged = {side: set(resolution.origin_tagged.get(side, ())) for side in sides}
    used = [side for side in sides if demoting[side] or tagged[side]]
    if not used:
        return []
    rekeyed = [
        cluster
        for cluster in resolution.index.vertices
        if isinstance(cluster.declaration, VertexEquivalence)
        and cluster.declaration.identity is not None
    ]
    declared = op.origins or {}

    def source(side: Side) -> str:
        if side in declared:
            return f"`origins.{side}`"
        return f"the {side} schema name"

    def names(side: Side) -> str:
        demoted = f"the {side} side's demoted keys"
        untagged = "local keys the op leaves untagged"
        if demoting[side] and tagged[side]:
            return f"names {demoted} and tags its {untagged}"
        if demoting[side]:
            return f"names {demoted}"
        return f"tags the {side} side's {untagged}"

    def classes(*intos: str) -> tuple[str, ...]:
        """The merged classes *intos*, in the index's order."""
        return tuple(
            subject("merged", cluster.into)
            for cluster in resolution.index.vertices
            if cluster.into in intos
        )

    out: list[MergeIdentityError] = []
    for side in used:
        origin = origins[side]
        problem = key_space_name_problem(origin)
        if problem is not None:
            out.append(
                MergeIdentityError(
                    f"merge_manifests: origin {origin!r} ({source(side)}) "
                    f"{names(side)}, but {problem}; set `origins.{side}` on the op",
                    check="origin",
                    subjects=classes(*demoting[side], *tagged[side]),
                )
            )
        authors = sorted(
            (f"{member_side}:{member}", cluster.into)
            for cluster in rekeyed
            for member_side in sides
            if (schema := manifests[member_side].graph_schema) is not None
            for member in cluster.members(member_side)
            if member in schema.core_schema.vertex_config.vertex_set
            and origin
            in {
                entry.name
                for entry in schema.core_schema.vertex_config[
                    member
                ].secondary_identities
            }
        )
        if authors:
            out.append(
                MergeIdentityError(
                    f"merge_manifests: origin {origin!r} ({source(side)}) "
                    f"{names(side)}, but "
                    f"{', '.join(author for author, _into in authors)} declare a "
                    f"secondary identity named {origin!r}; set `origins.{side}` "
                    "on the op",
                    check="origin",
                    subjects=classes(*(into for _author, into in authors)),
                )
            )
    if len(used) == 2 and origins["left"] == origins["right"]:
        out.append(
            MergeIdentityError(
                f"merge_manifests: both sides' origin is {origins['left']!r}, "
                "so the keys it names or tags on each side would share one "
                "name; set `origins` on the op",
                check="origin",
                subjects=classes(
                    *(into for side in sides for into in demoting[side] | tagged[side])
                ),
            )
        )
    return out

plan_demoted_key_names(into, members, key_spaces, origins, authored)

Name the demoted keys of the merged class into; pure.

members are its demoted members in order, as (side, member, fields, key): fields in the union's names (after the key-space prefix), key the canonical fields before it. A member's key space is its entry in key_spaces, else its side's origin. authored are the secondaries the members declare, by name, in the union's field names.

A field set authored declares answers to that declaration and its name. Otherwise the secondary is named by its key space, one per key space; a key space contributing two field sets names each <space>__<key fields joined by __>. A field set is shared by the members of one key space and by members of different sides; the first member to declare it fixes its column order. Refused: two key spaces of one side on one field set, authored or not -- their ids would share a lookup column -- and a <space>__<fields> name an authored secondary takes. A key space named like an authored secondary is :func:key_space_refusals' and :func:origin_refusals' to report.

Source code in graflo/architecture/evolution/merge_refusals.py

def plan_demoted_key_names(
    into: str,
    members: Sequence[tuple[Side, str, tuple[str, ...], tuple[str, ...]]],
    key_spaces: Mapping[tuple[Side, str], str],
    origins: Mapping[Side, str],
    authored: Mapping[str, Collection[str]],
) -> DemotedKeyPlan:
    """Name the demoted keys of the merged class *into*; pure.

    *members* are its demoted members in order, as ``(side, member, fields,
    key)``: *fields* in the union's names (after the key-space prefix), *key*
    the canonical fields before it. A member's key space is its entry in
    *key_spaces*, else its side's origin. *authored* are the secondaries the
    members declare, by name, in the union's field names.

    A field set *authored* declares answers to that declaration and its name.
    Otherwise the secondary is named by its key space, one per key space; a
    key space contributing two field sets names each ``<space>__<key fields
    joined by __>``. A field set is shared by the members of one key space
    and by members of different sides; the first member to declare it fixes
    its column order. Refused: two key spaces of one side on one field set,
    authored or not -- their ids would share a lookup column -- and a
    ``<space>__<fields>`` name an authored secondary takes.
    A key space named like an authored secondary is
    :func:`key_space_refusals`' and :func:`origin_refusals`' to report.
    """
    authored_by_fields = {frozenset(fields): name for name, fields in authored.items()}

    def space(side: Side, member: str) -> str:
        return key_spaces.get((side, member), origins[side])

    refusals: list[MergeIdentityError] = []
    claims: dict[tuple[Side, frozenset[str]], dict[str, list[str]]] = {}
    field_sets: dict[str, set[frozenset[str]]] = {}
    for side, member, fields, _key in members:
        claims.setdefault((side, frozenset(fields)), {}).setdefault(
            space(side, member), []
        ).append(member)
        if frozenset(fields) not in authored_by_fields:
            field_sets.setdefault(space(side, member), set()).add(frozenset(fields))
    for (side, fields), by_space in claims.items():
        if len(by_space) < 2:
            continue
        detail = ", ".join(
            f"{side}:{member} ({name!r})"
            for name, claimants in by_space.items()
            for member in claimants
        )
        refusals.append(
            MergeIdentityError(
                f"merge_manifests: key space: {detail} demote the key "
                f"{sorted(fields)} onto {into!r} from different key spaces, so "
                "their ids would share one lookup column and an id of one would "
                "find a node of another; tag the members' `local_key` alike "
                "(one key space), or do not list the key as a property branch "
                "of `identity`",
                check="key-space",
                subjects=(subject("merged", into),),
            )
        )

    by_fields = dict(authored_by_fields)
    names: dict[tuple[Side, str], str] = {}
    additions: list[SecondaryIdentity] = []
    for side, member, fields, key in members:
        if frozenset(fields) not in by_fields:
            name = space(side, member)
            if len(field_sets[name]) > 1:
                name = "__".join((name, *key))
                if name in authored:
                    refusals.append(
                        MergeIdentityError(
                            f"merge_manifests: key space: the key {side}:{member} "
                            f"demotes on {into!r} would be the secondary identity "
                            f"{name!r}, which a member already declares over "
                            f"{sorted(authored[name])}; rename that secondary, or "
                            f"set `origins.{side}` on the op, or tag the member's "
                            "`local_key` otherwise",
                            check="key-space",
                            subjects=(subject("merged", into),),
                        )
                    )
            entry = SecondaryIdentity(name=name, fields=list(fields))
            by_fields[frozenset(fields)] = name
            additions.append(entry)
        names[(side, member)] = by_fields[frozenset(fields)]
    return DemotedKeyPlan(names=names, additions=additions, refusals=refusals)