Documentation
¶
Overview ¶
Package bulkimport builds a labelled property graph from a stream of node and edge records, at bulk-loader speed, so that the result can be published as a store snapshot.
Why it exists ¶
The round-3 comparative audit measured GoGraph's Cypher write path at 35 m 33 s on a dataset Memgraph loads in 977 ms and Neo4j in 2.39 s — a 2184x deficit that dominates any first evaluation of the module — while store/bulk ingested the same edge volume in tens of milliseconds and was unreachable from anything a user could call. See docs/design-bulk-import.md for the spike that measured the alternatives (rmp #2177, #2178).
Why not store/bulk ¶
store/bulk is deliberately untouched. It ingests adjacency only — its record is (src, dst, weight) with no labels and no properties — and it emits a Tier-2 csrfile, which is exactly what its consumers (bench/ldbc, bench/rmat) want. Extending it to carry labels and properties would reimplement, in a second place, storage that lpg.Graph already owns and that is already tested.
The spike measured the cost of that reuse: driving lpg.Graph inside one adjacency commit window reaches 2.72 M edges/s against store/bulk's 3.92 M edges/s for adjacency alone — 44 % more time to carry a label and a property on every node and every edge.
This package measures 2.068 M edges/s (BenchmarkImport_LabelsAndProperties, ±2 % over 6 runs), 24 % below the spike's figure, and the gap is deliberate. The spike used the fused AddEdgeLabeledWithProperty; this package uses the HANDLE API — AddEdgeH, then SetEdgeLabelByHandle and SetEdgePropertyByHandle — because in a multigraph a pair may carry several edges and a pair-addressed write would silently overwrite the first edge's type and properties with the second's. 24 % is the price of not corrupting parallel edges, and it is worth paying: at 2.068 M edges/s the audit's 200 000-edge dataset builds in ~97 ms, so the whole import including publication is still ~150 ms — 6.5x faster than Memgraph's 977 ms, 16x faster than Neo4j's 2.39 s, and four orders of magnitude faster than the 35 m 33 s Cypher write path.
Concurrency ¶
A Builder is NOT safe for concurrent use. It holds an adjacency commit window open for its whole life, which is an exclusive-build mode: it assumes no concurrent reader and no concurrent writer on the graph it is building. Build on one goroutine, call Builder.Finish, and only then share the result.
Index ¶
- Variables
- type Builder
- func (b *Builder[W]) AddEdge(e Edge[W]) error
- func (b *Builder[W]) AddEdges(es []Edge[W]) error
- func (b *Builder[W]) AddNode(n Node) error
- func (b *Builder[W]) AddNodes(ns []Node) error
- func (b *Builder[W]) Finish() (Stats, error)
- func (b *Builder[W]) Graph() *lpg.Graph[string, W]
- func (b *Builder[W]) Stats() Stats
- type Edge
- type Node
- type Options
- type PublishResult
- type Stats
Constants ¶
This section is empty.
Variables ¶
var ErrFinished = errors.New("bulkimport: builder already finished")
ErrFinished is returned by every ingest method once Builder.Finish has run.
var ErrNotFinished = errors.New("bulkimport: builder has not been finished")
ErrNotFinished reports that Publish was given a Builder whose Builder.Finish has not run, so its adjacency commit window is still open.
var ErrStoreNotEmpty = errors.New("bulkimport: target store directory is not empty")
ErrStoreNotEmpty reports that the target directory already holds files, so it cannot be imported into. See Publish for why this is refused rather than merged.
Functions ¶
This section is empty.
Types ¶
type Builder ¶
type Builder[W any] struct { // contains filtered or unexported fields }
Builder accumulates node and edge records into an lpg.Graph.
The graph is built inside ONE adjacency commit window, opened at construction and closed by Builder.Finish. That is the same exclusive-build mode WAL replay uses, and snapshot recovery since rmp #2170: within a window a shard's slot array is cloned once on first touch and mutated in place thereafter, instead of once per edge. Outside a window the same import would cost O(edges x shard size).
Builder is NOT safe for concurrent use. See the package documentation.
func (*Builder[W]) AddEdge ¶
AddEdge ingests one edge record. Both endpoints must already exist; an unknown endpoint is an error rather than an implicit node creation, so a mistyped key cannot silently produce a bare node.
The edge is created through the HANDLE API (lpg.Graph.AddEdgeH) and its type and properties are attached to that handle. That is what makes parallel edges correct: in a multigraph, addressing an edge by (src, dst) alone is ambiguous, so a second edge between the same pair would otherwise overwrite the first one's type and properties.
func (*Builder[W]) AddEdges ¶
AddEdges ingests a batch of edge records, stopping at the first error.
func (*Builder[W]) AddNode ¶
AddNode ingests one node record, creating the node on first sight and merging labels and properties into it on any later sight of the same key.
func (*Builder[W]) AddNodes ¶
AddNodes ingests a batch of node records, stopping at the first error.
func (*Builder[W]) Finish ¶
Finish closes the adjacency commit window and returns what was ingested. It must be called exactly once, before Builder.Graph is used: the window's close is what freezes each touched shard's builder, so a graph handed out before it could still be mutated in place under a reader.
Finish is idempotent in the sense that a second call returns ErrFinished rather than closing the window twice.
func (*Builder[W]) Graph ¶
Graph returns the built graph. It is only valid after Builder.Finish; before that it returns nil, because the adjacency commit window is still open and the shards' builders are still mutable.
type Edge ¶
type Edge[W any] struct { Properties map[string]lpg.PropertyValue Src string Dst string // Type is the relationship type. Empty means an untyped edge. Type string Weight W }
Edge is one edge record: its endpoints, its weight, its relationship type, and its properties.
Src and Dst must have been added with Builder.AddNode first. That is a deliberate precondition rather than an implicit create: a typo in an edge file would otherwise silently produce a labelless, propertyless node, which is the class of silent-wrong-result the audit's correctness findings were about.
type Node ¶
type Node struct {
// Properties are set in map-iteration order, which is unspecified. That is
// safe because each key is written once, so no ordering can change the result.
Properties map[string]lpg.PropertyValue
Key string
Labels []string
}
Node is one node record: its natural key, its labels, and its properties.
Labels and Properties may both be empty; a node with neither is still created, which matters because an edge endpoint must exist. Duplicate keys are idempotent: the second record adds its labels and properties to the node the first created rather than failing, which is what a CSV split across files needs.
type Options ¶
type Options struct {
// Directed selects a directed graph. openCypher requires directed
// relationships, so a Cypher-facing import wants true.
Directed bool
// Multigraph allows parallel edges between the same pair. openCypher's data
// model is a multigraph, so a Cypher-facing import wants true; without it a
// second edge between an existing pair fails.
Multigraph bool
// ExpectNodes, when > 0, pre-sizes the interning table to that cardinality.
// It is a pure capacity hint with no effect on the result.
ExpectNodes int
}
Options configures a Builder.
type PublishResult ¶
type PublishResult struct {
// SnapshotDir is the published directory, <storeDir>/snapshot.
SnapshotDir string
// Stats are the builder's ingest counts, carried through for convenience.
Stats Stats
}
PublishResult reports what was written.
func ImportInto ¶
func ImportInto[W any]( ctx context.Context, storeDir string, opts Options, nodes []Node, edges []Edge[W], ) (PublishResult, error)
ImportInto is the one-call form: it builds a graph from nodes and edges and publishes it into storeDir. It is the entry point most callers want; use Builder plus Publish directly when the records must be streamed rather than held in slices.
The contract is Publish's in full: storeDir must be absent or empty, the import is atomic as a whole and is not a transaction, and it is concurrent with nothing.
func Publish ¶
Publish writes b's graph into storeDir as the store's snapshot, so that recovery.Open(storeDir) reconstructs it.
b must already have been finished with Builder.Finish; publishing a graph whose commit window is still open would hand out shards that are still mutable in place.
The directory must be empty ¶
storeDir must not exist, or must exist and be empty. A non-empty directory is refused with ErrStoreNotEmpty, and the check is enforced here rather than left to documentation, because the failure it prevents is silent corruption rather than an error: this path writes NO write-ahead log, so if the directory already held one, recovery would replay that WAL ON TOP of the freshly published snapshot. That is not a merge of old and new data — it is the old log's operations applied to an unrelated graph.
A store that must absorb bulk data into existing content uses the ordinary transactional write path. This one builds a store; it does not extend one.
What is atomic ¶
The whole import. snapshot.WriteSnapshotFullCtx assembles the snapshot under <storeDir>/snapshot.tmp and renames it to <storeDir>/snapshot on success, and a rename within a directory is atomic, so at every instant the store either has no snapshot or has a complete one. There is no state in which a reader can observe part of the imported graph.
A crash before the rename leaves the assembly directory behind. Recovery neither opens it — it is not the name recovery reads — nor keeps it: recovery removes a stale <snapshot>.tmp on open. So a crashed import leaves a store that looks exactly as it did before the import started.
What is NOT atomic, and must not be read as such ¶
- This is not a transaction. It has no transaction id, appends no WAL record, participates in no isolation level, and cannot be rolled back once published. Undoing it means deleting the directory.
- There is no per-record durability acknowledgement and no resumption point. Nothing is durable until the rename; a caller that streams ten million edges and crashes at nine million has no partial result to resume from and re-runs the import.
- It is concurrent with nothing. No reader, no writer and no checkpointer may touch storeDir during the import. That is why this is an offline Go call and not a Cypher clause: publishing a snapshot under a live server would race the checkpointer and invalidate open readers' view.
Durability of the published bytes rests on the snapshot writer's existing fsync discipline — each file fsynced, then the parent directory — which is the same protocol the checkpointer uses and which the crash-injection battery already exercises. This adds no new durability mechanism, deliberately.
type Stats ¶
type Stats struct {
// Nodes is the number of DISTINCT node keys created.
Nodes int
// Edges is the number of edge records ingested.
Edges int
// NodeRecords is the number of node records consumed, which exceeds Nodes
// when the input repeats a key.
NodeRecords int
}
Stats reports what an import ingested.
Source Files