Documentation
¶
Overview ¶
Package bulk implements the bulk-loading path that bypasses the transactional WAL stack and writes a Tier 2 csrfile directly from a stream of edges.
Bulk loading is the high-throughput equivalent of running many txn.Commit calls back-to-back. The loader pipes edges into an in-memory adjacency list and then writes the resulting CSR through csrfile.WriteToFile; a future revision will introduce an external k-way merge sort for graphs that exceed memory.
Pre-sizing ¶
When Options.MaxRows > 0 the loader treats the cap as a capacity hint and pre-sizes the adjacency builder and its interning table via adjlist.AdjList.Reserve, eliminating most slice/map re-growths on the ingest hot path. Pre-sizing is a pure allocation optimisation: it never changes which NodeID a key receives nor the order of edges in the resulting CSR.
Partitioned parallel ingest ¶
For large directed loads the loader can build the adjacency in parallel across bounded goroutines (see Options.Parallel) while producing a result that is byte-for-byte identical to the sequential loader. Determinism is guaranteed by a two-phase scheme: a serial first phase assigns every NodeID in input order (reproducing the sequential interning order exactly by construction), then a parallel second phase builds adjacency partitioned by the source node's Mapper shard so that partitions write to disjoint shards with no contention and each source keeps its edges in input order. Undirected and simple-graph loads, and loads below a small threshold, use the sequential path because mirror/dedup edges would cross partition boundaries; the result is identical either way.
Example ¶
Example bulk-loads a small graph: edges are streamed through a Loader that bypasses the transactional WAL stack and writes a Tier 2 csrfile directly. Finalise returns the row count and the in-memory CSR; the file is then reopened to confirm it landed on disk.
package main
import (
"fmt"
"os"
"path/filepath"
"github.com/FlavioCFOliveira/GoGraph/store/bulk"
"github.com/FlavioCFOliveira/GoGraph/store/csrfile"
)
func main() {
dir, err := os.MkdirTemp("", "bulk-example")
if err != nil {
panic(err)
}
defer func() { _ = os.RemoveAll(dir) }()
out := filepath.Join(dir, "graph.csr")
l := bulk.New(bulk.Options{OutputPath: out, Directed: true})
// Feed edges one at a time and in a batch; both paths funnel into
// the same in-memory adjacency list.
if err := l.Add(bulk.Edge{Src: "a", Dst: "b", Weight: 1}); err != nil {
panic(err)
}
if err := l.AddBatch([]bulk.Edge{
{Src: "b", Dst: "c", Weight: 2},
{Src: "c", Dst: "a", Weight: 3},
}); err != nil {
panic(err)
}
// Finalise flushes the accumulated edges to the csrfile and returns
// the row count plus the resulting CSR snapshot.
rows, c, err := l.Finalise()
if err != nil {
panic(err)
}
fmt.Printf("rows=%d csr-order=%d csr-size=%d\n", rows, c.Order(), c.Size())
// Reopen the written file to confirm the bulk load is durable.
r, err := csrfile.Open(out)
if err != nil {
panic(err)
}
defer func() { _ = r.Close() }()
fmt.Printf("on-disk edges=%d\n", r.Header().NEdges)
}
Output: rows=3 csr-order=3 csr-size=3 on-disk edges=3
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
var ErrTooManyRows = errors.New("bulk: row cap exceeded")
ErrTooManyRows is returned by Loader.Add, Loader.AddBatch, and Loader.Drain when the configured Options.MaxRows cap is exceeded.
Functions ¶
This section is empty.
Types ¶
type Loader ¶
type Loader struct {
// contains filtered or unexported fields
}
Loader streams edges through an in-memory adjacency list and writes the resulting Tier 2 csrfile when Loader.Finalise runs.
Loader is not safe for concurrent use; callers that wish to parallelise ingestion either set Options.Parallel (the loader then fans out internally during Finalise) or partition the edge stream upstream and call separate Loaders, then merge.
func New ¶
New returns a fresh Loader.
When opts.ExpectNodes > 0 the interning table is pre-sized to that node estimate (a pure, determinism-neutral capacity hint; see Options.ExpectNodes). When opts.Parallel is set the loader buffers edges for the parallel build performed by Loader.Finalise, pre-sizing the edge buffer to opts.MaxRows when that cap is known.
func (*Loader) Add ¶
Add ingests one edge. Returns ErrTooManyRows when the row cap is exceeded.
func (*Loader) AddBatch ¶
AddBatch ingests a contiguous batch of edges. Returns ErrTooManyRows on the first edge that would cross the cap; edges accepted before that point remain ingested.
func (*Loader) Drain ¶
Drain consumes from ch until it is closed or ctx is cancelled. Returns the number of edges drained and any error from the input channel (ErrTooManyRows when the row cap is exceeded).
func (*Loader) Finalise ¶
Finalise builds the CSR from the accumulated edges and writes it to opts.OutputPath as a csrfile. Returns the row count, the resulting CSR (for chaining into search/extern), and any error.
When Options.Parallel is set and the buffered load is a large directed graph, Finalise builds the adjacency in parallel; the resulting CSR and csrfile are byte-for-byte identical to the sequential build. The csrfile is always published atomically and durably by csrfile.WriteToFile (tmp + fsync + rename + parent fsync): the parallel build completes fully in memory before the single publication, so a crash mid-build leaves no partial csrfile.
type Options ¶
type Options struct {
// OutputPath is the destination csrfile.
OutputPath string
// Directed selects the adjacency-list configuration.
Directed bool
// Multigraph allows parallel edges in the loaded graph.
Multigraph bool
// MaxRows, when > 0, caps the number of edge records the loader
// will ingest. Add / AddBatch / Drain return [ErrTooManyRows]
// on the row that crosses the cap. Default (0) is unbounded.
//
// MaxRows additionally serves as a capacity hint: when set, the
// adjacency builder and its interning table are pre-sized to it so
// the ingest hot path incurs far fewer slice/map re-growths.
MaxRows int
// ExpectNodes, when > 0, is the caller's estimate of the number of
// DISTINCT nodes the load will produce. It pre-sizes the interning
// table (the Mapper) to that cardinality, eliminating most of the
// map/slice re-growth the first-encounter Intern path incurs. Unlike
// MaxRows (an edge count), ExpectNodes sizes the node-indexed
// interning structures correctly, so it reduces allocations without
// the over-allocation an edge-count hint would cause. It is a pure,
// determinism-neutral capacity hint. Leave it 0 when the node count
// is unknown.
ExpectNodes int
// Parallel selects partitioned-parallel ingest for large directed
// loads. The default (false) always uses the deterministic
// sequential build. When true, the loader buffers edges and builds
// the adjacency across up to GOMAXPROCS (capped at an internal
// bound) goroutines during [Loader.Finalise], producing a result
// byte-for-byte identical to the sequential build. Parallelism is
// only engaged for directed loads at or above an internal edge-count
// threshold; smaller, undirected, or simple-graph loads transparently
// fall back to the sequential build.
Parallel bool
}
Options configures the Loader.
Source Files