graflo.db.cypher.rel_merge

Relationship MERGE map fragments for parallel edges (property-graph backends).

Functions:

partition_by_absent_merge_props(rows, prop_names, *, props_of, ends_of, union=())

Group edge rows by the identity properties each one lacks.

prop_names are the properties the edge's identity names beside its endpoints. Cypher refuses MERGE on a null property, so a row missing one (absent or None) cannot use the property MERGE. As on PostgreSQL (NULLS NOT DISTINCT), absence is part of the key: such a row matches an existing edge lacking the same properties. Rows lacking some are deduplicated on their endpoints and present merge values, later properties winning, because :func:rel_absent_upsert_clause cannot see edges created earlier in its own batch.

With union fields the complete group is deduplicated as well: one row per edge carries the union of its rows' members, so no MERGE has to see a row written earlier in its own batch.

Returns:

Type	Description
list[tuple[tuple[str, ...], list[Any]]]	(absent names, rows) pairs; the group lacking nothing comes first.

Source code in graflo/db/cypher/rel_merge.py

def partition_by_absent_merge_props(
    rows: Sequence[Any],
    prop_names: Sequence[str],
    *,
    props_of: Callable[[Any], Mapping[str, Any] | None],
    ends_of: Callable[[Any], tuple[Any, Any]],
    union: Collection[str] = (),
) -> list[tuple[tuple[str, ...], list[Any]]]:
    """Group edge rows by the identity properties each one lacks.

    *prop_names* are the properties the edge's identity names beside its
    endpoints. Cypher refuses ``MERGE`` on a null property, so a row missing
    one (absent or ``None``) cannot use the property ``MERGE``. As on
    PostgreSQL (``NULLS NOT DISTINCT``), absence is part of the key: such a row
    matches an existing edge lacking the same properties. Rows lacking some are
    deduplicated on their endpoints and present merge values, later properties
    winning, because :func:`rel_absent_upsert_clause` cannot see edges created
    earlier in its own batch.

    With *union* fields the complete group is deduplicated as well: one row per
    edge carries the union of its rows' members, so no ``MERGE`` has to see a row
    written earlier in its own batch.

    Returns:
        ``(absent names, rows)`` pairs; the group lacking nothing comes first.
    """
    keys = _normalized_prop_names(prop_names)
    groups: dict[tuple[str, ...], list[Any]] = {}
    position: dict[tuple[tuple[str, ...], str], int] = {}
    for row in rows:
        props = props_of(row) or {}
        absent = tuple(k for k in keys if props.get(k) is None)
        group = groups.setdefault(absent, [])
        if not absent and not union:
            group.append(row)
            continue
        source, target = ends_of(row)
        ident = _key_text(
            [source, target, [props.get(k) for k in keys if k not in absent]]
        )
        at = position.get((absent, ident))
        if at is None:
            position[(absent, ident)] = len(group)
            group.append(row)
        else:
            group[at] = _merged_row(group[at], row, props_of, union)
    return sorted(groups.items(), key=lambda item: len(item[0]))

rel_absent_upsert_clause(relation_name, prop_names, absent, *, props_expr, source='source', target='target', union=(), union_expr='row[3]')

Upsert a relationship whose identity properties absent are missing.

Matches an existing relation_name edge between source and target whose present identity properties equal the row's and whose absent ones are unset, and updates it; creates one when none matches. Follows a MATCH binding source, target and row. union fields are read from union_expr and unioned with what the edge holds.

Source code in graflo/db/cypher/rel_merge.py

def rel_absent_upsert_clause(
    relation_name: str,
    prop_names: Sequence[str],
    absent: Sequence[str],
    *,
    props_expr: str,
    source: str = "source",
    target: str = "target",
    union: Sequence[str] = (),
    union_expr: str = "row[3]",
) -> str:
    """Upsert a relationship whose identity properties *absent* are missing.

    Matches an existing ``relation_name`` edge between *source* and *target*
    whose present identity properties equal the row's and whose *absent* ones are
    unset, and updates it; creates one when none matches. Follows a ``MATCH``
    binding *source*, *target* and ``row``. *union* fields are read from
    *union_expr* and unioned with what the edge holds.
    """
    rel = cypher_name(relation_name)
    conditions = []
    for key in _normalized_prop_names(prop_names):
        bk = cypher_map_key(key)
        if key in absent:
            conditions.append(f"existing.{bk} IS NULL")
        else:
            conditions.append(
                f"existing.{bk} = {props_expr}[{cypher_string_literal(key)}]"
            )
    return (
        f"OPTIONAL MATCH ({source})-[existing:{rel}]->({target})\n"
        f"WHERE {' AND '.join(conditions)}\n"
        "FOREACH (_ IN CASE WHEN existing IS NULL THEN [1] ELSE [] END |\n"
        f"    CREATE ({source})-[r:{rel}]->({target}) SET r += {props_expr}"
        f"{_union_set('r', union, union_expr)})\n"
        "FOREACH (_ IN CASE WHEN existing IS NULL THEN [] ELSE [1] END |\n"
        f"    SET existing += {props_expr}"
        f"{_union_set('existing', union, union_expr)})"
    )

rel_merge_props_map_from_row_index(prop_names, *, row_index=2)

Build `k`: row[n]['k'], ... for MERGE relationship properties.

Matches batches shaped as row = [source_doc, target_doc, props] (Neo4j, FalkorDB-style row[2]).

Source code in graflo/db/cypher/rel_merge.py

def rel_merge_props_map_from_row_index(
    prop_names: Sequence[str], *, row_index: int = 2
) -> str:
    """Build `` `k`: row[n]['k'], ... `` for MERGE relationship properties.

    Matches batches shaped as ``row`` = ``[source_doc, target_doc, props]`` (Neo4j,
    FalkorDB-style ``row[2]``).
    """
    row_access = f"row[{row_index}]"
    parts: list[str] = []
    for key in _normalized_prop_names(prop_names):
        bk = cypher_map_key(key)
        lit = cypher_string_literal(key)
        parts.append(f"{bk}: {row_access}[{lit}]")
    return ", ".join(parts)

rel_merge_props_map_from_row_props(prop_names, *, props_expr='row.props')

Build `k`: row.props['k'], ... (Memgraph-style batch rows).

Source code in graflo/db/cypher/rel_merge.py

def rel_merge_props_map_from_row_props(
    prop_names: Sequence[str], *, props_expr: str = "row.props"
) -> str:
    """Build `` `k`: row.props['k'], ... `` (Memgraph-style batch rows)."""
    parts: list[str] = []
    for key in _normalized_prop_names(prop_names):
        bk = cypher_map_key(key)
        lit = cypher_string_literal(key)
        parts.append(f"{bk}: {props_expr}[{lit}]")
    return ", ".join(parts)

rel_union_rows(rows, union)

[source, target, props] rows as [source, target, replaced, unioned].

The union fields move out of row[2] into row[3], so r += row[2] can never overwrite a stored list (:func:rel_upsert_clause).

Source code in graflo/db/cypher/rel_merge.py

def rel_union_rows(rows: Sequence[Any], union: Sequence[str]) -> list[list[Any]]:
    """``[source, target, props]`` rows as ``[source, target, replaced, unioned]``.

    The union fields move out of ``row[2]`` into ``row[3]``, so ``r += row[2]``
    can never overwrite a stored list (:func:`rel_upsert_clause`).
    """
    out: list[list[Any]] = []
    for row in rows:
        props = (row[2] if len(row) > 2 else None) or {}
        out.append(
            [
                row[0],
                row[1],
                {k: v for k, v in props.items() if k not in union},
                {k: props[k] for k in union if k in props},
            ]
        )
    return out

rel_upsert_clause(relation_name, prop_names, absent, *, union=(), props_expr='row[2]', union_expr='row[3]')

Upsert one group of :func:partition_by_absent_merge_props rows.

MERGE on the endpoints (and prop_names when the group has them all), or :func:rel_absent_upsert_clause when it lacks some. Follows a MATCH binding source, target and row. union fields are read from union_expr (:func:rel_union_rows) and written as the union of the stored list and the row's.

Source code in graflo/db/cypher/rel_merge.py

def rel_upsert_clause(
    relation_name: str,
    prop_names: Sequence[str],
    absent: Sequence[str],
    *,
    union: Sequence[str] = (),
    props_expr: str = "row[2]",
    union_expr: str = "row[3]",
) -> str:
    """Upsert one group of :func:`partition_by_absent_merge_props` rows.

    ``MERGE`` on the endpoints (and *prop_names* when the group has them all),
    or :func:`rel_absent_upsert_clause` when it lacks some. Follows a ``MATCH``
    binding ``source``, ``target`` and ``row``. *union* fields are read from
    *union_expr* (:func:`rel_union_rows`) and written as the union of the
    stored list and the row's.
    """
    if absent:
        return rel_absent_upsert_clause(
            relation_name,
            prop_names,
            absent,
            props_expr=props_expr,
            union=union,
            union_expr=union_expr,
        )
    rel = cypher_name(relation_name)
    if prop_names:
        map_clause = rel_merge_props_map_from_row_index(prop_names)
        pattern = f"MERGE (source)-[r:{rel} {{{map_clause}}}]->(target)"
    else:
        pattern = f"MERGE (source)-[r:{rel}]->(target)"
    return f"{pattern}\nSET r += {props_expr}" + _union_set("r", union, union_expr)