Bulk loading#

Writing elements one CREATE at a time pays a statement, and usually a round trip, for each one. The bulk loaders use PostgreSQL’s COPY in its binary format instead, which streams every element in one operation: about thirty times a loop of one CREATE per row inside a single transaction, and about half again on top of the fastest single UNWIND ... CREATE statement.

Method

Writes

load_vertices(label, properties)

one vertex per property map

load_edges(label, edges)

one edge per (start, end, properties)

identity_map(label, key)

nothing: it reads the identities edges need

upsert_vertices(label, key, rows)

only the vertices whose key is not there yet, and optionally refreshes the rest

upsert_edges(label, edges)

only the edges whose pair of endpoints is not there yet

load_vertex_frame(label, frame), load_edge_frame(label, frame)

from an Arrow, pandas or polars frame

All of them write to the graph the session has selected, or to graph=, and run on the connection like any statement: on a connection that is not in autocommit, nothing is visible until you commit.

Before loading#

The label must exist. COPY writes into the label’s table and cannot create one. Create the labels first, with CREATE VLABEL and CREATE ELABEL or with ensure_labels():

conn.execute("CREATE VLABEL studio")
conn.execute("CREATE ELABEL produced")

Name the label as the server stores it. The name you pass is quoted, so it is taken literally, and the server stores an unquoted name in lower case: CREATE VLABEL Studio makes studio, and load_vertices("Studio", ...) then fails with UndefinedTable. Pass "studio".

Vertices#

>>> conn.load_vertices("studio", [
...     {"name": "Warner Bros."},
...     {"name": "Legendary"},
...     {"name": "Lucasfilm"},
... ])
3

load_vertices() takes any iterable of mappings, a generator included, consumes it as it goes, and returns how many vertices it wrote. Each mapping becomes one vertex’s property map, encoded the way a dict parameter is: NaN and the infinities are refused, and a datetime becomes its ISO text.

No identities are sent. Each new vertex gets its identity from the label, exactly as a CREATE would give it, so nothing on the client has to generate or coordinate identities.

The whole load is one statement. If any row fails, for example on a uniqueness constraint, none of them is written.

Edges#

An edge needs the identities of the two vertices it joins. identity_map() reads them, keyed by a property:

>>> studios = conn.identity_map("studio", "name")
>>> studios
{'Warner Bros.': GraphId(labid=8, locid=1), 'Legendary': GraphId(labid=8, locid=2), 'Lucasfilm': GraphId(labid=8, locid=3)}
>>> movies = conn.identity_map("movie", "title")
>>> conn.load_edges("produced", [
...     (studios["Warner Bros."], movies["The Matrix"], {"year": 1999}),
...     (studios["Legendary"], movies["Interstellar"], None),
... ])
2
>>> conn.execute_query(
...     "MATCH (s:studio)-[:produced]->(m:movie) RETURN s.name, m.title ORDER BY s.name"
... ).records
[('Legendary', 'Interstellar'), ('Warner Bros.', 'The Matrix')]

Each edge is (start, end, properties), where start and end are GraphIds and properties is a mapping or None. The endpoints are not checked: an edge to an identity no vertex has is written as given, so take the identities from the graph, as above.

identity_map()#

It returns {key: GraphId}, with every key as text: a key stored as the number 1 is found under "1". It refuses rather than guesses when the key does not identify one element, because a map that silently kept one of two elements would attach every edge of both to whichever it kept:

>>> conn.identity_map("person", "born")
Traceback (most recent call last):
  ...
ValueError: 'born' does not identify an element of 'person': 5 value(s) shared by more than one element (['1947', '1964', '1965', '1967', '1970']). An edge resolved through this map would land on the wrong element or on none

An element that has no such property at all is refused the same way.

By default it reads the whole label. Every property of an element lives in one jsonb column, so reading one key reads the whole property map of every element, and on a label whose elements carry large maps, such as embeddings, that is most of the cost of a load. Two things make it cheap:

  • Name the keys you need, with keys=, on a key something keeps unique (a unique property index or a uniqueness constraint). Each key is then one index lookup, and only the keys found are returned. Without a uniqueness guarantee, keys= is ignored and the whole label is read.

  • Give the key a column of its own, below. The key is then read from that column without touching the map.

Measured on 20,000 elements that each carry a 1536-dimension embedding: the whole label takes about 630 milliseconds whatever is indexed, against 3 milliseconds from a key with a column of its own.

Writing what may already be there#

A copy only creates. Loading the same source twice makes a second vertex for each one the graph already holds. upsert_vertices() reads which keys are there and copies only the rows whose key is not:

>>> conn.execute("CREATE UNIQUE PROPERTY INDEX ON studio (name)")
<agensgraph.connection.Cursor ...>
>>> conn.upsert_vertices("studio", "name", [
...     {"name": "Pixar", "founded": 1986},
...     {"name": "Lucasfilm", "founded": 1971},
... ])
UpsertCounts(inserted=1, updated=0)

Pixar was missing and was written; Lucasfilm was there and was left alone. With on_existing="update", an existing element is updated instead, in one statement for all of them. The update merges: a property the row does not mention keeps its value.

>>> conn.upsert_vertices(
...     "studio", "name",
...     [{"name": "Lucasfilm", "founded": 1971}, {"name": "A24"}],
...     on_existing="update",
... )
UpsertCounts(inserted=1, updated=1)
>>> conn.execute_query(
...     "MATCH (s:studio) WHERE s.name IN ['Lucasfilm', 'A24']"
...     " RETURN s.name, s.founded ORDER BY s.name"
... ).records
[('A24', None), ('Lucasfilm', 1971)]

It returns UpsertCounts(inserted, updated), from agensgraph.bulk. It is a copy and at most one statement, so it is much faster than a MERGE per row.

The key must be unique, and something must keep it so. Without a unique property index or a uniqueness constraint on the key, the call is refused:

ValueError: nothing makes 'name' unique on 'studio', so two writers merging on it would each create
an element rather than find one. Add a unique property index or a uniqueness constraint, or pass
require_unique=False to accept the hazard

This is not caution for its own sake. Eight concurrent writers merging on twenty-five shared keys, with only a plain index, left twenty-seven elements for the twenty-five keys on a live server. With a unique index, a writer that loses the race fails with a unique violation instead, which is the one to retry, classified with merging=True. Pass require_unique=False only for a graph that cannot have the index and a load nothing else runs at the same time as.

Two more things to know:

  • A key given twice in one call is not merged. Both rows go to the copy, and the unique index refuses the second. Remove duplicates from a batch first.

  • It is several statements: a read, a copy and possibly an update. Run it in a transaction if it must succeed or fail as a whole.

Edges#

upsert_edges() keys an edge on the pair of vertices it joins, since that pair is what an edge is. What keeps one edge per pair has to be an index over the edge table’s start and end columns, which is plain SQL, in that order:

>>> conn.execute('CREATE UNIQUE INDEX produced_pair ON moviekb.produced (start, "end")')
<agensgraph.connection.Cursor ...>
>>> conn.upsert_edges("produced", [
...     (studios["Warner Bros."], movies["The Matrix"], {"year": 1999}),
...     (studios["Warner Bros."], movies["The Matrix Reloaded"], {"year": 2003}),
...     (studios["Warner Bros."], movies["The Matrix Reloaded"], {"year": 2003}),
... ])
UpsertCounts(inserted=1, updated=0)

The first pair was already there and the second was not. The third repeats the second, and is written once: unlike vertices, a pair given twice in one call is only copied once. on_existing="update" merges new properties into edges that exist.

It decides for itself how to find out which pairs exist: by asking about the pairs in the batch, or, for a batch as large as the label, by reading the label’s edges, whichever the planner’s estimate of the label’s size says is cheaper. With the pair index in place, eight concurrent writers over the same pairs left one edge per pair, and the losers failed with a unique violation.

Give the key a column of its own#

If a label is loaded more than once, keep its key in a column of its own, from AgensGraph 2.18. This is a schema decision rather than a driver setting, and it matters more than anything else on this page:

conn.execute("CREATE VLABEL doc (key text GENERATED)")
conn.execute("CREATE UNIQUE PROPERTY INDEX ON doc (key)")

Or declared, as ensure_labels() writes it. A property in a column of its own is read without reading the rest of the property map, and it is the only form an index can answer from without visiting the table. On 20,000 elements with 1536-dimension embeddings, reading every key took 731 milliseconds and 60,183 buffers from the map, and 3 milliseconds and 163 buffers from the column.

Adding the column to a label that already holds elements rewrites the label’s table. A read of the whole element, RETURN n, still goes through the map, so the column speeds up reads of that key, not everything.

From a DataFrame#

A frame goes in directly, without becoming Python objects on the way. It needs the arrow extra, and takes an Arrow table or record batches, a pandas DataFrame, a polars DataFrame, or anything else Arrow can read:

>>> import pyarrow
>>> frame = pyarrow.table({"name": ["Amblin", "Carolco"], "founded": [1981, 1976]})
>>> conn.load_vertex_frame("studio", frame)
2

Each column becomes a property of that name. For edges, two columns hold the endpoints, start and end by default (start= and end= name others), as packed integers (GraphId.packed) or as text such as '8.1', and the remaining columns become properties:

>>> amblin = conn.identity_map("studio", "name", keys=["Amblin"])["Amblin"]
>>> conn.load_edge_frame("produced", pyarrow.table({
...     "start": [amblin.packed],
...     "end": [movies["Jurassic Park"].packed],
...     "credit": ["production company"],
... }))
1

It is faster than handing the same rows to load_vertices() as mappings, because the rows never become Python objects. Most of a load is the server’s own work, so what falls is mostly the client’s share.

A null cell is stored as a property whose value is JSON null, not left out as a None in a mapping is. Drop or fill nulls first if the difference matters to your queries.

Embeddings in bulk#

load_vertices() encodes a property map as JSON, and a Vector is deliberately not JSON, so pass an embedding to the bulk loaders as a plain list of floats. Kept in a column of its own of type vector(n), it reads back as a Vector all the same.

Memory during a large read#

Graph values are not tracked by Python’s cycle collector, so a large result costs the collector little. Two helpers remain for the rest of an application:

import agensgraph

# once at startup: move what already exists out of the collector's way
agensgraph.freeze_after_import()

with agensgraph.paused_collection():    # no collection while a large result is read
    rows = conn.execute_query("MATCH (m:movie) RETURN m").records

Measured on a 200,000-vertex read, pausing collection is worth about three percent; it matters more in an application whose other objects the collector would otherwise walk.