Skip to content

graflo.db.cypher.bulk

Create-only Cypher for a bulk load, and the counts that verify it.

In a bulk load every row is new, so nodes and relationships are created with no MERGE: nothing is looked for but a relationship's two endpoints. Rows are $rows: node property maps, or [source keys, target keys, properties]. A null property is not stored, as an upsert's += removes it.

Attributes

Run = Callable[[str], list[dict[str, Any]]] module-attribute

Classes

Functions:

bulk_counts(run)

Nodes per label and relationships per (source label, type, target label).

A relationship type with one source and one target label is counted from per-label counts, which Neo4j answers from its count store; a type with several is counted by a scan of that type alone.

Source code in graflo/db/cypher/bulk.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def bulk_counts(run: Run) -> BulkCounts:
    """Nodes per label and relationships per (source label, type, target label).

    A relationship type with one source and one target label is counted from
    per-label counts, which Neo4j answers from its count store; a type with
    several is counted by a scan of that type alone.
    """
    labels = [r["label"] for r in run("CALL db.labels() YIELD label RETURN label")]
    vertices = {
        label: int(_single(run(f"MATCH (n:{cypher_name(label)}) RETURN count(n) AS c")))
        for label in labels
    }
    types = [
        r["relationshipType"]
        for r in run(
            "CALL db.relationshipTypes() YIELD relationshipType RETURN relationshipType"
        )
    ]
    edges: dict[StoredEdgeId, int] = {}
    for rel in types:
        r = cypher_name(rel)
        sources = {
            label: int(
                _single(
                    run(
                        f"MATCH (:{cypher_name(label)})-[r:{r}]->() RETURN count(r) AS c"
                    )
                )
            )
            for label in labels
        }
        targets = {
            label: int(
                _single(
                    run(
                        f"MATCH ()-[r:{r}]->(:{cypher_name(label)}) RETURN count(r) AS c"
                    )
                )
            )
            for label in labels
        }
        sources = {k: v for k, v in sources.items() if v}
        targets = {k: v for k, v in targets.items() if v}
        if len(sources) == 1 and len(targets) == 1:
            ((source, count),) = sources.items()
            (target,) = targets
            edges[(source, rel, target)] = count
            continue
        for row in run(
            f"MATCH (s)-[r:{r}]->(t) "
            "RETURN head(labels(s)) AS s, head(labels(t)) AS t, count(r) AS c"
        ):
            edges[(row["s"], rel, row["t"])] = int(row["c"])
    return BulkCounts(vertices={k: v for k, v in vertices.items() if v}, edges=edges)

create_nodes_query(label)

Create one label node per row.

Source code in graflo/db/cypher/bulk.py 
21
22
23
def create_nodes_query(label: str) -> str:
    """Create one ``label`` node per row."""
    return f"UNWIND $rows AS row CREATE (n:{cypher_name(label)}) SET n = row"

create_relationships_query(source_label, target_label, relation, match_keys_source, match_keys_target)

Create one relationship per row between endpoints found by their keys.

A row whose endpoint is missing creates nothing; the load's count check reports it.

Source code in graflo/db/cypher/bulk.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def create_relationships_query(
    source_label: str,
    target_label: str,
    relation: str,
    match_keys_source: Sequence[str],
    match_keys_target: Sequence[str],
) -> str:
    """Create one relationship per row between endpoints found by their keys.

    A row whose endpoint is missing creates nothing; the load's count check
    reports it.
    """
    return (
        "UNWIND $rows AS row "
        f"MATCH (source:{cypher_name(source_label)} "
        f"{{{_key_map(match_keys_source, 0)}}}) "
        f"MATCH (target:{cypher_name(target_label)} "
        f"{{{_key_map(match_keys_target, 1)}}}) "
        f"CREATE (source)-[r:{cypher_name(relation)}]->(target) SET r = row[2]"
    )

scanned_counts(run)

Nodes per label and relationships per (source label, type, target label), by a scan.

For a server without db.labels() or a count store; one pass over the nodes and one over the relationships.

Source code in graflo/db/cypher/bulk.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def scanned_counts(run: Run) -> BulkCounts:
    """Nodes per label and relationships per (source label, type, target label), by a scan.

    For a server without ``db.labels()`` or a count store; one pass over the
    nodes and one over the relationships.
    """
    vertices = {
        row["label"]: int(row["c"])
        for row in run("MATCH (n) RETURN head(labels(n)) AS label, count(n) AS c")
    }
    edges: dict[StoredEdgeId, int] = {
        (row["s"], row["r"], row["t"]): int(row["c"])
        for row in run(
            "MATCH (s)-[r]->(t) RETURN head(labels(s)) AS s, type(r) AS r, "
            "head(labels(t)) AS t, count(r) AS c"
        )
    }
    return BulkCounts(vertices=vertices, edges=edges)