Batching Large Spatial Joins Safely

An index-probe join that runs beautifully over ten thousand driving rows will terminate the transaction over ten million. The query is unchanged and the plan is unchanged; what changed is that the whole result now has to be held until commit, and the transaction memory limit — or the heap, if nobody set one — arrives first. Splitting the work into committed chunks fixes that, and introduces two problems of its own: the run is no longer atomic, so a failure halfway leaves the graph partly updated, and the chunks have to be chosen so that resuming does not redo or skip work. This page does all three: bounded transactions, a resumable cursor, and a chunking key that keeps each batch’s probes spatially local.

Prerequisites & Versions

Ordinary Cypher plus a checkpoint the job owns.

Requirement Minimum version Install
Python 3.11 —
neo4j (async driver) 5.20 pip install "neo4j>=5.20"
Neo4j Server 5.15 native point, POINT INDEX

Implementation

import asyncio
from dataclasses import dataclass

from neo4j import AsyncGraphDatabase

# One batch = one transaction. The keyset predicate on the chunk key is what
# makes it resumable: a restart continues strictly past the last committed
# chunk rather than re-scanning from the beginning.
JOIN_CHUNK = """
MATCH (e:PickupEvent)
WHERE e.geocell > $after_cell
   OR (e.geocell = $after_cell AND e.id > $after_id)
WITH e ORDER BY e.geocell, e.id LIMIT $size
// The probe: one bounding-box seek per driving row, then the exact clip.
CALL (e) {
  WITH e
  MATCH (n:RoadNode)
  WHERE n.location.latitude  >= e.min_lat AND n.location.latitude  <= e.max_lat
    AND n.location.longitude >= e.min_lon AND n.location.longitude <= e.max_lon
  WITH e, n, point.distance(e.location, n.location) AS metres
  WHERE metres <= $radius_m
  RETURN n, metres ORDER BY metres LIMIT 1
}
MERGE (e)-[r:SNAPPED_TO]->(n)
SET r.metres = metres, r.run_id = $run_id
RETURN e.geocell AS last_cell, e.id AS last_id, count(r) AS written
ORDER BY last_cell DESC, last_id DESC
LIMIT 1
"""

CHECKPOINT = """
MERGE (c:JoinCheckpoint {job: $job})
SET c.after_cell = $cell, c.after_id = $id,
    c.written = coalesce(c.written, 0) + $written, c.updated_at = datetime()
"""

READ_CHECKPOINT = """
MATCH (c:JoinCheckpoint {job: $job})
RETURN c.after_cell AS cell, c.after_id AS id, c.written AS written
"""


@dataclass(frozen=True)
class Progress:
    batches: int
    written: int
    last_cell: int
    last_id: str


class BatchedJoin:
    def __init__(self, uri: str, auth: tuple[str, str], job: str) -> None:
        self._driver = AsyncGraphDatabase.driver(uri, auth=auth)
        self._job = job

    async def close(self) -> None:
        await self._driver.close()

    async def run(self, radius_m: float = 60.0, size: int = 2_000,
                  run_id: str = "run-1") -> Progress:
        async with self._driver.session() as session:
            record = await (await session.run(READ_CHECKPOINT, job=self._job)).single()
            cell = int(record["cell"]) if record else -1
            last_id = str(record["id"]) if record else ""
            written = int(record["written"] or 0) if record else 0
            batches = 0

            while True:
                result = await session.run(
                    JOIN_CHUNK, after_cell=cell, after_id=last_id,
                    size=size, radius_m=radius_m, run_id=run_id,
                )
                row = await result.single()
                if row is None or row["written"] == 0:
                    break        # nothing left past the cursor

                cell, last_id = int(row["last_cell"]), str(row["last_id"])
                written += int(row["written"])
                batches += 1
                # Checkpoint AFTER the batch commits. The other order records
                # progress that a crash would then have discarded, and the
                # resume skips work that was never done.
                await session.run(
                    CHECKPOINT, job=self._job, cell=cell, id=last_id,
                    written=int(row["written"]),
                )

        return Progress(batches=batches, written=written,
                        last_cell=cell, last_id=last_id)

How It Works

Chunking on a spatial key keeps each batch’s probes local. Ordering by geocell rather than by id means the driving rows in one batch are geographically adjacent, so their bounding-box seeks land on overlapping index pages and the same pages serve the whole batch. Chunking by id instead scatters each batch across the country, and every probe is a fresh page fault — the same work, several times the wall-clock, and a page cache thrashing against itself.

The cursor is a keyset, not an offset. SKIP would re-produce and discard every already-processed row on each batch, making the run quadratic; the compound predicate on (geocell, id) seeks directly past the last committed position. It also makes the run resumable across process restarts for free, since the cursor lives in the checkpoint rather than in memory.

The checkpoint is written after the batch commits, and that ordering is the whole safety argument. Recording progress first and committing second means a crash in between leaves a checkpoint claiming work that was rolled back, and the resume skips it permanently. Committing first means a crash leaves work done but unrecorded, so the resume redoes one batch — which is harmless, because the MERGE is idempotent.

Chunking by id scatters the probes; chunking by cell keeps them together Two batches of two thousand driving rows over the same dataset. Ordered by id, the rows in one batch are scattered across the whole country, so each bounding-box probe lands on a different part of the index and almost every one is a page fault. Ordered by grid cell, the rows in one batch are geographically adjacent, so their probes land on overlapping index pages and the pages loaded for the first probe serve most of the rest. The rows processed and the result written are identical; only the page-cache behaviour differs, and on a large run that is most of the wall-clock. One batch of 2,000 driving rows, two orderings ORDER BY id every probe a fresh index page ORDER BY geocell, id one page set serves the whole batch Identical rows, identical output, identical plan. The right-hand ordering costs a sort the left-hand one does not — and pays for it many times over in page faults it never takes.

Common Failure Patterns

1. Checkpointing before the commit. The ordering looks arbitrary and is not: a crash between a recorded checkpoint and a rolled-back transaction leaves a permanent hole in the output, and nothing detects it because the run reports completion. Commit, then record — and accept that a crash costs one redone batch.

2. Batching without a stable order. A LIMIT with no ORDER BY returns an arbitrary subset, so consecutive batches can overlap and miss rows in the same run. The order must be total — the geocell, id pair, exactly as a keyset cursor needs a tiebreak — or the run is not reproducible.

3. Treating a partial run as a failure. Once the work is chunked it is no longer atomic, and that is a deliberate trade rather than a defect. What matters is that the intermediate state is usable: the run_id stamped on each relationship makes it possible to tell which rows this run produced, so a downstream consumer can filter to a completed run or an operator can clean up an abandoned one.

// What a resumed or abandoned run left behind.
MATCH ()-[r:SNAPPED_TO]->()
RETURN r.run_id AS run, count(r) AS rows, min(r.metres) AS closest
ORDER BY rows DESC;

Performance Notes

Batch size trades transaction memory against round-trip overhead, and both ends of the range are bad:

$$T_{\text{total}} \approx \frac{N}{B} \cdot t_{\text{rtt}} + N \cdot t_{\text{probe}}, \qquad M_{\text{tx}} \approx B \cdot m_{\text{row}}$$

Small batches make the first term dominate — a hundred-row batch over ten million rows is a hundred thousand round trips. Large batches push $M_{\text{tx}}$ toward the transaction memory limit, which is where this whole exercise started. A few thousand rows is the usual landing point, and the right way to choose it is to measure rows-per-second across a range and take the setting on the near side of the plateau.

The spatial ordering has a second benefit worth noting: it makes the run’s progress predictable. Because batches sweep the grid in order, the job moves through geography rather than through an opaque id space, so “how far through is it” has a meaningful answer and a stalled batch can be attributed to a specific region — usually the densest one, where each probe returns the most candidates.

Where the join must not interfere with live traffic, the same chunking gives a natural throttle: a short sleep between batches caps the write rate without changing anything else, and because the run is resumable it can be stopped and restarted around peak hours with no bookkeeping beyond the checkpoint that already exists.

Batch size against throughput and transaction memory Two curves against batch size. Throughput rises steeply from very small batches as the per-round-trip overhead is amortised, flattens through a broad plateau between roughly one and eight thousand rows, and then falls as transactions become large enough to pressure memory. Transaction memory rises linearly with batch size throughout and crosses the configured limit at around twenty thousand rows. The usable setting is on the near side of the throughput plateau and well below the memory line — which on this workload is anywhere between two and six thousand. Throughput and transaction memory against batch size 1001k5k20k100k rows per batch throughput transaction memory configured limit usable range The plateau is broad, so this is a setting to measure once and leave alone rather than one to tune continuously.

Two further considerations decide whether this is safe to run against a live system rather than only against a maintenance window.

The first is what the job does to the page cache it shares with everything else. A join sweeping the full grid pulls the entire driving label and a large part of the road index through the cache, and the pages it loads displace whatever the latency-sensitive workload had resident. The spatial ordering helps — each batch’s pages are reused within the batch rather than scattered — but it does not change the total volume, and on a shared instance the routing endpoint’s p95 will move while the job runs. Where that matters, the answer is to run against a replica, or to accept it and schedule accordingly; what does not work is hoping a job that reads the whole graph will be invisible.

The second is what happens when the job is interrupted permanently rather than temporarily. A run stopped halfway leaves the graph in a state where some driving rows are snapped and some are not, and a consumer reading it cannot distinguish “not snapped yet” from “no road within tolerance”. Stamping the run id on the relationships is half the answer; the other half is stamping progress on the driving rows themselves — a snap_run_id set as each batch commits — so an unsnapped row that was processed is distinguishable from one that never was. Without that, resuming after an abandoned run requires re-processing everything to find out what was missed, which defeats the resumability the checkpoint was built for.

It is also worth deciding up front whether re-running the job over already-snapped rows should update them. Road geometry moves between imports, so a row snapped six months ago may now have a nearer segment; but re-snapping everything on every run turns an incremental job back into a full one. The usual compromise is to re-snap rows whose local road geometry changed — which the import already knows, since it wrote those changes — and to leave the rest alone.

This guide is part of Spatial Join Techniques, within Cypher Spatial Queries & Pathfinding Patterns.

Siblings