graflo.architecture.evolution.physical

Typing and physical-profile evolution ops.

Property types and secondary indexes were visible to the schema differ long before anything could author them. These ops close that half of the loop; they do not touch a live database, which stays the job of the migration executor.

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

Functions:

apply_add_edge_indexes(manifest, op)

Author secondary indexes on edge physical specs.

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

def apply_add_edge_indexes(manifest: GraphManifest, op: AddEdgeIndexesOp) -> None:
    """Author secondary indexes on edge physical specs."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("add_edge_indexes requires graph_schema")

    known_edges = {edge.edge_id for edge in schema.core_schema.edge_config.edges}
    for entry in op.edges:
        edge_id = (entry.source, entry.target, entry.relation)
        if edge_id not in known_edges:
            raise ValueError(f"add_edge_indexes: unknown edge: {edge_id}")
        if not entry.indexes:
            raise ValueError(f"add_edge_indexes: edge {edge_id} lists no indexes")

        spec = _edge_spec_for(manifest, entry.physical_key())
        if spec is None:
            raise ValueError(
                f"add_edge_indexes: no physical spec for {entry.physical_key()}"
            )
        existing = {frozenset(index.fields) for index in spec.indexes}
        additions: list[Index] = []
        for index in entry.indexes:
            if frozenset(index.fields) in existing:
                raise ValueError(
                    f"add_edge_indexes: {entry.physical_key()} already indexes "
                    f"{index.fields}"
                )
            existing.add(frozenset(index.fields))
            additions.append(index.model_copy(deep=True))
        spec.indexes = [*spec.indexes, *additions]

    schema.finish_init()

apply_add_vertex_indexes(manifest, op)

Author secondary indexes on vertices.

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

def apply_add_vertex_indexes(manifest: GraphManifest, op: AddVertexIndexesOp) -> None:
    """Author secondary indexes on vertices."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("add_vertex_indexes requires graph_schema")

    vertex_config = schema.core_schema.vertex_config
    unknown = sorted(set(op.indexes) - vertex_config.vertex_set)
    if unknown:
        raise ValueError(f"add_vertex_indexes: unknown vertices: {unknown}")

    for vertex_name, indexes in op.indexes.items():
        declared = {
            field.name
            for vertex in vertex_config.vertices
            if vertex.name == vertex_name
            for field in vertex.properties
        }
        for index in indexes:
            missing = [name for name in index.fields if name not in declared]
            if missing:
                raise ValueError(
                    f"add_vertex_indexes: vertex '{vertex_name}' does not declare "
                    f"{missing}"
                )
            schema.db_profile.add_vertex_index(vertex_name, index.model_copy(deep=True))

    schema.finish_init()

apply_change_field_types(manifest, op)

Set the logical type of existing vertex and edge properties.

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

def apply_change_field_types(manifest: GraphManifest, op: ChangeFieldTypesOp) -> None:
    """Set the logical type of existing vertex and edge properties."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("change_field_types requires graph_schema")

    vertex_config = schema.core_schema.vertex_config
    unknown = sorted(set(op.vertices) - vertex_config.vertex_set)
    if unknown:
        raise ValueError(f"change_field_types: unknown vertices: {unknown}")

    for vertex in vertex_config.vertices:
        changes = op.vertices.get(vertex.name)
        if not changes:
            continue
        declared = {field.name for field in vertex.properties}
        missing = sorted(set(changes) - declared)
        if missing:
            raise ValueError(
                f"change_field_types: vertex '{vertex.name}' does not declare {missing}"
            )
        identity_fields = set(vertex.identity) | set(vertex.hash_identity_properties)
        for field_name, spec in changes.items():
            if spec.type == FieldType.LIST and field_name in identity_fields:
                raise ValueError(
                    f"change_field_types: vertex '{vertex.name}' field "
                    f"'{field_name}' participates in the identity and cannot become "
                    "a LIST"
                )
        new_properties = []
        for field in vertex.properties:
            spec = changes.get(field.name)
            if spec is None:
                new_properties.append(field)
                continue
            retyped = _retyped(field, spec)
            _assert_supported(manifest, retyped)
            new_properties.append(retyped)
        vertex.properties = new_properties

    if op.edges or op.per_edge:
        relations = {
            edge.relation
            for edge in schema.core_schema.edge_config.edges
            if edge.relation
        }
        unknown_relations = sorted(set(op.edges) - relations)
        if unknown_relations:
            raise ValueError(
                f"change_field_types: unknown relations: {unknown_relations}"
            )

        # A relation can join several endpoint pairs, each declaring its own
        # properties: a field is retyped on every edge of the relation that
        # declares it, and refused only when none does.
        edges = schema.core_schema.edge_config.edges
        for relation, changes in op.edges.items():
            declared = {
                field.name
                for edge in edges
                if edge.relation == relation
                for field in edge.properties
            }
            missing = sorted(set(changes) - declared)
            if missing:
                raise ValueError(
                    f"change_field_types: edge '{relation}' does not declare {missing}"
                )
        on_edge = {edge.edge_id: {f.name for f in edge.properties} for edge in edges}
        for entry in op.per_edge:
            if entry.edge_id() not in on_edge:
                raise ValueError(f"change_field_types: unknown edge: {entry.edge_id()}")
            missing = sorted(set(entry.fields) - on_edge[entry.edge_id()])
            if missing:
                raise ValueError(
                    f"change_field_types: edge {entry.edge_id()} does not declare "
                    f"{missing}"
                )

        for edge in edges:
            changes = op.edge_changes_for(edge.edge_id)
            if not changes:
                continue
            declared = {field.name for field in edge.properties}
            identity_tokens = {token for key in edge.identities for token in key}
            for field_name, spec in changes.items():
                if field_name not in declared:
                    continue
                if spec.type == FieldType.LIST and field_name in identity_tokens:
                    raise ValueError(
                        f"change_field_types: edge '{edge.relation}' field "
                        f"'{field_name}' participates in an identity key and "
                        "cannot become a LIST"
                    )
            new_properties = []
            for field in edge.properties:
                spec = changes.get(field.name)
                if spec is None:
                    new_properties.append(field)
                    continue
                retyped = _retyped(field, spec)
                _assert_supported(manifest, retyped)
                new_properties.append(retyped)
            edge.properties = new_properties

    schema.finish_init()

apply_remove_edge_indexes(manifest, op)

Withdraw authored indexes from edge physical specs.

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

def apply_remove_edge_indexes(manifest: GraphManifest, op: RemoveEdgeIndexesOp) -> None:
    """Withdraw authored indexes from edge physical specs."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("remove_edge_indexes requires graph_schema")

    for entry in op.edges:
        if not entry.fields:
            raise ValueError(
                f"remove_edge_indexes: edge {entry.physical_key()} lists no fields"
            )
        spec = _edge_spec_for(manifest, entry.physical_key())
        if spec is None:
            raise ValueError(
                f"remove_edge_indexes: no physical spec for {entry.physical_key()}"
            )
        doomed = {frozenset(fields) for fields in entry.fields}
        present = {frozenset(index.fields) for index in spec.indexes}
        unmatched = sorted(sorted(item) for item in doomed - present)
        if unmatched:
            raise ValueError(
                f"remove_edge_indexes: {entry.physical_key()} has no index on "
                f"{unmatched}"
            )
        spec.indexes = [
            index for index in spec.indexes if frozenset(index.fields) not in doomed
        ]

    schema.finish_init()

apply_remove_vertex_indexes(manifest, op)

Withdraw authored vertex indexes, refusing to touch derived ones.

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

def apply_remove_vertex_indexes(
    manifest: GraphManifest, op: RemoveVertexIndexesOp
) -> None:
    """Withdraw authored vertex indexes, refusing to touch derived ones."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("remove_vertex_indexes requires graph_schema")

    unknown = sorted(set(op.indexes) - schema.core_schema.vertex_config.vertex_set)
    if unknown:
        raise ValueError(f"remove_vertex_indexes: unknown vertices: {unknown}")

    for vertex_name, field_lists in op.indexes.items():
        derived = _derived_secondary_identity_field_sets(manifest, vertex_name)
        doomed = {frozenset(fields) for fields in field_lists}
        collision = sorted(sorted(entry) for entry in doomed & derived)
        if collision:
            raise ValueError(
                f"remove_vertex_indexes: vertex '{vertex_name}' indexes {collision} "
                "are derived from secondary_identities and would be re-registered by "
                "the next finish_init; use RemoveSecondaryIdentitiesOp instead"
            )

        existing = schema.db_profile.vertex_indexes.get(vertex_name, [])
        present = {frozenset(index.fields) for index in existing}
        unmatched = sorted(sorted(entry) for entry in doomed - present)
        if unmatched:
            raise ValueError(
                f"remove_vertex_indexes: vertex '{vertex_name}' has no index on "
                f"{unmatched}"
            )
        schema.db_profile.vertex_indexes[vertex_name] = [
            index for index in existing if frozenset(index.fields) not in doomed
        ]

    # finish_init prunes any entry this emptied — an empty list and an absent
    # key mean the same thing but hash differently.
    schema.finish_init()

apply_set_bindings(manifest, op)

Replace the manifest's bindings block.

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

def apply_set_bindings(manifest: GraphManifest, op: SetBindingsOp) -> None:
    """Replace the manifest's bindings block."""
    manifest.bindings = (
        op.bindings.model_copy(deep=True) if op.bindings is not None else None
    )

apply_set_db_profile(manifest, op)

Replace the schema's database profile.

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

def apply_set_db_profile(manifest: GraphManifest, op: SetDbProfileOp) -> None:
    """Replace the schema's database profile."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("set_db_profile requires graph_schema")

    schema.db_profile = op.profile.model_copy(deep=True)
    schema.finish_init()

apply_set_edge_directed(manifest, op)

Set the directed flag on selected edges.

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

def apply_set_edge_directed(manifest: GraphManifest, op: SetEdgeDirectedOp) -> None:
    """Set the ``directed`` flag on selected edges."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("set_edge_directed requires graph_schema")

    by_edge_id = {edge.edge_id: edge for edge in schema.core_schema.edge_config.edges}
    unknown = sorted(
        str(selector.edge_id())
        for selector in op.edges
        if selector.edge_id() not in by_edge_id
    )
    if unknown:
        raise ValueError(f"set_edge_directed: unknown edges: {unknown}")

    for selector in op.edges:
        by_edge_id[selector.edge_id()].directed = op.directed

    schema.finish_init()

apply_set_native_inverses(manifest, op)

Add relations to, or withdraw them from, db_profile.native_inverses.

The rules (declared pair, not symmetric, no explicit inverse edges, one physical edge type, TigerGraph) are the schema's own and are checked by :meth:Schema.finish_init; this op only names itself in the refusal.

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

def apply_set_native_inverses(manifest: GraphManifest, op: SetNativeInversesOp) -> None:
    """Add relations to, or withdraw them from, ``db_profile.native_inverses``.

    The rules (declared pair, not symmetric, no explicit inverse edges, one
    physical edge type, TigerGraph) are the schema's own and are checked by
    :meth:`Schema.finish_init`; this op only names itself in the refusal.
    """
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("set_native_inverses requires graph_schema")

    known = {
        edge.relation
        for edge in schema.core_schema.edge_config.edges
        if edge.relation is not None
    }
    unknown = sorted(set(op.relations) - known)
    if unknown:
        raise ValueError(f"set_native_inverses: unknown relations: {unknown}")

    profile = schema.db_profile
    current = set(profile.native_inverses)
    selected = set(op.relations)
    profile.native_inverses = sorted(
        current | selected if op.enabled else current - selected
    )
    try:
        schema.finish_init()
    except ValueError as exc:
        raise ValueError(f"set_native_inverses: {exc}") from exc