graflo.architecture.evolution.merge

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

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

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 none of the three ways to resolve it were declared: an explicit identity on the VertexEquivalence, a PropertyEquivalence(identity=True) flag, or an identity_alignments entry for the merged class. The alternative -- silently taking the union of both field-sets as the new identity -- produces a natural key no record fully carries.

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.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 none of the three ways to resolve it were declared: an
    explicit ``identity`` on the ``VertexEquivalence``, a
    ``PropertyEquivalence(identity=True)`` flag, or an ``identity_alignments``
    entry for the merged class. The alternative -- silently taking the
    union of both field-sets as the new identity -- produces a natural key no
    record fully carries.

    ``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: str = "", subjects: tuple[str, ...] = ()
    ) -> None:
        # The only refusal carrying a default check: every raise site here 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.py

def __init__(
    self, message: str, *, check: str = "", subjects: tuple[str, ...] = ()
) -> None:
    # The only refusal carrying a default check: every raise site here 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.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`.
    """

Functions:

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.

When op.identity_alignments is non-empty, the merged union is further rewritten by the fundamental ops emitted from each alignment (see :func:~graflo.architecture.evolution.alignment.alignment_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``.

    When ``op.identity_alignments`` is non-empty, the merged union is further
    rewritten by the fundamental ops emitted from each alignment (see
    :func:`~graflo.architecture.evolution.alignment.alignment_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.
    """
    if not isinstance(op, MergeManifestsOp):
        raise TypeError(f"merge_manifests expects MergeManifestsOp, got {type(op)!r}")
    maps = _coerce_side_maps(canonical_maps)

    out_left = left.model_copy(deep=True)
    out_right = right.model_copy(deep=True)

    left_schema = _schema_of(out_left)
    right_schema = _schema_of(out_right)
    if op.target_namespace is not None:
        # The op's choice supersedes both sides' declarations, so a
        # disagreement between them is resolved rather than refused.
        for schema in (left_schema, right_schema):
            if schema is not None:
                schema.db_profile.target_namespace = None
    for side, schema in (("left", left_schema), ("right", right_schema)):
        if schema is None:
            logger.info(
                "merge_manifests: %s manifest carries no schema block; "
                "the merged schema comes from the other side alone",
                side,
            )

    # Raw ClusterConflictError here (not wrapped): a merge op whose own
    # declarations conflict is broken regardless of any canonical map.
    resolution = resolve_clusters(
        op, left=out_left, right=out_right, canonical_maps=maps
    )
    index = resolution.index
    side_maps = resolution.side_maps

    member_keys, member_property_names = _capture_all_member_state(
        index, left_schema, right_schema, side_maps
    )

    left_resource_names: set[str] = set()
    if out_left.ingestion_model is not None:
        left_resource_names = {r.name for r in out_left.ingestion_model.resources}
    _apply_right_resource_policy(out_right, op, left_resource_names)

    # The sides as the identity alignments see them: resources carry the names
    # the union will use, and every cluster member still exists as its own
    # class. The per-side lowering below merges the members in place, after
    # which no manifest can say which router key produced which member.
    sides = {
        "left": out_left.model_copy(deep=True),
        "right": out_right.model_copy(deep=True),
    }

    for manifest, side in ((out_left, "left"), (out_right, "right")):
        apply_manifest_ops_inplace(manifest, canonicalize_ops(side_maps[side]))

    _apply_right_schema_collision_policy(out_left, out_right, op, index)

    alignment_labels = {alignment.vertex for alignment in op.identity_alignments}
    # Three-way, deliberately: a fabricated empty Schema would not be neutral.
    # Its `DatabaseProfile` declares nothing, but it would still fold a
    # fabricated label into the merged name, and a side that declared the
    # default flavor could not be told from one that never spoke.
    post_left = _schema_of(out_left)
    post_right = _schema_of(out_right)
    retire_ops: list[ManifestOp] = []
    composed_schema: Schema | None
    if post_left is not None and post_right is not None:
        composed_schema, retire_ops = _union_schema(
            post_left,
            post_right,
            index,
            member_keys,
            member_property_names,
            alignment_labels,
        )
    elif post_left is not None:
        composed_schema = post_left.model_copy(deep=True)
    elif post_right is not None:
        composed_schema = post_right.model_copy(deep=True)
    else:
        composed_schema = None
    if composed_schema is not None:
        _assert_no_canonical_split(composed_schema)
    composed_ingestion = _concat_ingestion(
        out_left.ingestion_model, out_right.ingestion_model
    )
    composed_bindings = _union_bindings(
        out_left.bindings, out_right.bindings, name_conflict=op.name_conflict
    )

    result = GraphManifest(
        graph_schema=composed_schema,
        ingestion_model=composed_ingestion,
        bindings=composed_bindings,
        metadata=_merge_manifest_metadata(left.metadata, right.metadata),
    )
    _bump_schema_version(result, bump_version)

    if retire_ops:
        apply_manifest_ops_inplace(result, retire_ops)

    if op.identity_alignments:
        result = _apply_identity_alignments(
            result,
            op,
            index=index,
            sides=sides,
            side_maps=side_maps,
            canonical_maps=[
                ("left", resolution.declared.left),
                ("right", resolution.declared.right),
            ],
            finish_init=False,
            strict_references=strict_references,
            dynamic_edge_feedback=dynamic_edge_feedback,
        )

    _apply_merge_naming(result, op)

    if finish_init:
        result.finish_init(
            strict_references=strict_references,
            dynamic_edge_feedback=dynamic_edge_feedback,
        )
    return result

side_identity_to_funnel(cluster, member_property_names)

Lower a :class:~graflo.architecture.evolution.ops.SideIdentity shorthand.

Each member supplies an ordered branch chain (its own override, or the side default); the chains are merged into one global branch order by topological sort over the "comes before" relation each chain implies, breaking ties by first appearance across members (in declaration order: every left member, then every right member). Two members that disagree on the relative order of two branches have no consistent global order and raise :class:MergeIdentityError.

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

def side_identity_to_funnel(
    cluster: Cluster,
    member_property_names: dict[tuple[Side, str], set[str]],
) -> IdentityFunnel:
    """Lower a :class:`~graflo.architecture.evolution.ops.SideIdentity` shorthand.

    Each member supplies an ordered branch chain (its own override, or the
    side default); the chains are merged into one global branch order by
    topological sort over the "comes before" relation each chain implies,
    breaking ties by first appearance across members (in declaration order:
    every left member, then every right member). Two members that disagree on
    the relative order of two branches have no consistent global order and
    raise :class:`MergeIdentityError`.
    """
    side_identity = cluster.declaration.identity
    assert isinstance(side_identity, SideIdentity)

    chains: dict[tuple[Side, str], list[tuple[str, ...]]] = {}
    for side in ("left", "right"):
        default = side_identity.left if side == "left" else side_identity.right
        overrides = {
            cluster.resolved(side, declared): chain
            for declared, chain in side_identity.members.items()
        }
        for member in cluster.members(side):
            raw = overrides.get(member, default)
            if raw is None:
                raise MergeIdentityError(
                    f"merge_manifests: cluster into {cluster.into!r} has no "
                    f"SideIdentity branch chain for {side}:{member} (no "
                    f"per-member override and no {side} default)"
                )
            chains[(side, member)] = [_branch_tuple(spec) for spec in raw]

    ordered_members: list[tuple[Side, str]] = [("left", m) for m in cluster.left] + [
        ("right", m) for m in cluster.right
    ]
    first_seen: dict[tuple[str, ...], int] = {}
    for key in ordered_members:
        for branch in chains[key]:
            first_seen.setdefault(branch, len(first_seen))

    indegree: dict[tuple[str, ...], int] = dict.fromkeys(first_seen, 0)
    successors: dict[tuple[str, ...], set[tuple[str, ...]]] = {
        b: set() for b in first_seen
    }
    for key in ordered_members:
        chain = chains[key]
        for a, b in pairwise(chain):
            if b not in successors[a]:
                successors[a].add(b)
                indegree[b] += 1

    remaining = dict(indegree)
    available = sorted(
        (b for b, deg in remaining.items() if deg == 0), key=first_seen.__getitem__
    )
    order: list[tuple[str, ...]] = []
    while available:
        node = available.pop(0)
        order.append(node)
        for succ in successors[node]:
            remaining[succ] -= 1
            if remaining[succ] == 0:
                available.append(succ)
        available.sort(key=first_seen.__getitem__)

    if len(order) != len(first_seen):
        raise MergeIdentityError(
            f"merge_manifests: cluster into {cluster.into!r} has an "
            "inconsistent SideIdentity branch order — two members disagree "
            "on the relative order of two branches, so no single global "
            "funnel order satisfies both"
        )

    for (side, member), chain in chains.items():
        declared = member_property_names.get((side, member), set())
        for branch in chain:
            missing = sorted(set(branch) - declared)
            if missing:
                raise MergeIdentityError(
                    f"merge_manifests: cluster into {cluster.into!r} "
                    f"SideIdentity branch {branch} on {side}:{member} "
                    f"references undeclared propert"
                    f"{'y' if len(missing) == 1 else 'ies'} {missing}"
                )

    return IdentityFunnel(
        branches=[
            IdentityBranch(id="_".join(fields), fields=list(fields)) for fields in order
        ]
    )