exec

package
v0.8.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 14, 2026 License: MIT Imports: 26 Imported by: 0

Documentation

Overview

Package exec implements the Volcano-style executor for the Cypher query engine. It defines the Operator interface, the Row/RowSlab data model, and the pipeline driver Drain.

Data model

A Row is a slice of expr.Value. A RowSlab is a bounded, pooled container of pre-allocated rows used to eliminate per-row heap allocations in the hot path.

Concurrency

RowSlab is NOT safe for concurrent use. Each goroutine must obtain its own slab from NewRowSlab or from a sync.Pool managed by the caller. The exported SlabPool provides a ready-to-use pool with default capacity.

Index

Examples

Constants

View Source 
const (
	VLEDirForward = 0
	VLEDirReverse = 1
)

VLEDirForward and VLEDirReverse are the direction-marker values stored at the third slot of each hop triple in the VarLengthExpand flat path list (stride VLEHopStride). A reverse marker tells the relationship hydrator to swap the storage endpoints when resolving the edge's per-instance type and properties, and tells the path renderer to emit `<-[…]-` (rmp #1685).

View Source 
const DefaultMaxDistinct = 10_000_000

DefaultMaxDistinct is the default upper bound on distinct rows tracked by the Distinct operator.

View Source 
const DefaultMaxEagerRows = 10_000_000

DefaultMaxEagerRows is the default upper bound on rows that Eager holds in memory. It matches the sibling pipeline-breaker caps (DefaultMaxSortRows, DefaultMaxDistinct).

View Source 
const DefaultMaxGroups = 1_000_000

DefaultMaxGroups is the default upper bound on distinct groups that EagerAggregation will hold in memory.

View Source 
const DefaultMaxSortRows = 10_000_000

DefaultMaxSortRows is the default upper bound on rows that Sort holds in memory.

View Source 
const DefaultMorselSize = 1024

DefaultMorselSize is the number of NodeIDs processed per worker goroutine per scheduling quantum. Sized to fill roughly one or two cache lines of work before touching a channel.

View Source 
const DefaultSlabCapacity = 4096

DefaultSlabCapacity is the default maximum number of rows a RowSlab holds before returning ErrSlabOverflow. It is sized to keep a typical pipeline batch within a few cache lines.

View Source 
const VLEHopStride = 3

VLEHopStride is the per-hop element count of the flat alternating path list that VarLengthExpand emits. The list is 

[srcNodeID, fwdPos0, dstNode0, dir0, fwdPos1, dstNode1, dir1, …]

so an N-hop path occupies 1 + VLEHopStride*N elements: the leading source node id, then one (forward edge position, destination node id, direction marker) triple per hop. The direction marker is VLEDirForward or VLEDirReverse. Readers in the cypher package (the path/relationship-list hydrators) and [appendExcludedFromValue] index the list with this stride; it is exported so those readers share the single source of truth (rmp #1685).

Before #1685 the stride was 2 ((edgePos, dst) pairs) and the edge position was synthetic for a reverse hop, which lost both the physical-edge identity and the traversal direction the hydrator needs for per-instance type and property reporting on multigraph parallel edges.

Variables

View Source 
var ErrAggMemoryExceeded = errors.New("exec: aggregation memory cap exceeded")

ErrAggMemoryExceeded is returned by EagerAggregation.Next when the number of distinct groups exceeds the configured maxGroups limit.

View Source 
var ErrConstraintAlreadyExists = errors.New("exec: constraint already exists")

ErrConstraintAlreadyExists is the sentinel returned (wrapped) when CREATE CONSTRAINT (without IF NOT EXISTS) names a constraint whose (kind, label, property) identity is already registered. It is the analogue of Neo4j's EquivalentSchemaRuleAlreadyExists / ConstraintAlreadyExists — a constraint- level fault surfaced instead of leaking the synthetic backing-index name.

View Source 
var ErrConstraintNameConflict = errors.New("exec: constraint name already in use")

ErrConstraintNameConflict is the sentinel returned (wrapped) when CREATE CONSTRAINT requests a name already held by a DIFFERENT constraint (a distinct kind/label/property). Neo4j requires constraint names to be unique across the database; this is the analogue of its ConstraintWithNameAlreadyExists.

View Source 
var ErrConstraintNotFound = errors.New("exec: constraint not found")

ErrConstraintNotFound is the sentinel returned (wrapped) when a DROP CONSTRAINT names a constraint that does not exist and IF EXISTS was not given. It is the fail-stop analogue of Neo4j's ConstraintDropFailed / "No such constraint": the drop reports a typed error rather than fail-silently claiming success.

View Source 
var ErrConstraintViolation = errors.New("exec: constraint violation")

ErrConstraintViolation is the sentinel returned (wrapped) by CheckSetProperty when a write would violate a constraint.

View Source 
var ErrDeleteNodeHasRelationships = errors.New("exec: cannot delete node with existing relationships; use DETACH DELETE")

ErrDeleteNodeHasRelationships is returned when DELETE is attempted on a node that still has one or more incident relationships. Use DETACH DELETE to remove the node together with its relationships.

View Source 
var ErrDistinctMemoryExceeded = errors.New("exec: distinct memory cap exceeded")

ErrDistinctMemoryExceeded is returned by Distinct.Next when the number of distinct rows seen exceeds the configured maxDistinct limit.

View Source 
var ErrEagerMemoryExceeded = errors.New("exec: eager memory cap exceeded")

ErrEagerMemoryExceeded is returned by Eager.Init when the drained child produces more than maxRows rows.

View Source 
var ErrHashJoinMemoryExceeded = errors.New("exec: hash join memory cap exceeded")

ErrHashJoinMemoryExceeded is returned by the HashJoin build phase when the estimated retained size of the build table exceeds the configured byte budget (#1841). The build table is otherwise unbounded, so without a budget a large build side could exhaust memory before any drain-level guard fires.

View Source 
var ErrIndexTypeMismatch = errors.New("exec: index type mismatch")

ErrIndexTypeMismatch is returned by NodeByIndexSeek.Init when the seek value's Kind is incompatible with the index's key type.

View Source 
var ErrNestedPropertyValue = errors.New("exec: InvalidPropertyType: a nested list or map is not a valid property value")

ErrNestedPropertyValue is returned when a property value is a nested collection — a list containing a list or a map. openCypher restricts a property value to a primitive or a flat list of primitives; a nested collection is InvalidPropertyType. Unlike a non-literal expression (which is deferred to a runtime PropsEvalFn), this sentinel is a hard, fail-stop error: the CREATE/MERGE literal builders re-raise it rather than silently skipping the property, so an invalid value is never stored (the storage layer cannot serialise a nested PropList) and never silently dropped (audit 2026-07-13 security F3).

View Source 
var ErrProjectionRowTooLarge = errors.New("exec: projection row memory cap exceeded")

ErrProjectionRowTooLarge is returned by Project.Next when the estimated size of a single assembled output row exceeds the configured per-row byte budget. It bounds the transient peak of one row's construction (e.g. a RETURN that projects several large list columns) to the ceiling plus one column, independent of the column count, so a query cannot OOM the process by compounding many big columns into one row (#1852).

View Source 
var ErrPropertyValueIsNull = errors.New("exec: property value is null (skip)")

ErrPropertyValueIsNull is the sentinel returned by [parsePropValue] when the value is the literal "null". By openCypher semantics, assigning null to a property removes it (or never sets it for a fresh node), so callers that catch this sentinel must skip the property entirely rather than surface a parse error.

View Source 
var ErrSchemaMismatch = errors.New("exec: union schema mismatch: column counts differ")

ErrSchemaMismatch is returned when the left and right operands of a UNION produce rows with different column counts.

View Source 
var ErrSlabOverflow = errors.New("exec: row slab overflow")

ErrSlabOverflow is returned by RowSlab.Alloc when the slab has reached its capacity limit. Callers must flush or reset the slab before continuing.

View Source 
var ErrSortMemoryExceeded = errors.New("exec: sort memory cap exceeded")

ErrSortMemoryExceeded is returned when Sort collects more than maxRows rows.

View Source 
var ErrUnwindNilChild = errors.New("exec: NewUnwind requires non-nil child Operator")

ErrUnwindNilChild is returned by NewUnwind when child is nil.

View Source 
var ErrUnwindNilListFn = errors.New("exec: NewUnwind requires non-nil listFn")

ErrUnwindNilListFn is returned by NewUnwind when listFn is nil.

View Source 
var ErrVarLenCapExceeded = errors.New("exec: variable-length expand safety cap exceeded")

ErrVarLenCapExceeded is returned when a VarLengthExpand exceeds its configured maximum edge traversal count — either the per-input-row cap or the aggregate per-query cap (see [defaultMaxEdgesTraversed] and [defaultMaxTotalEdgesTraversed]).

Functions

func MergeActionEvalKey added in v0.8.0 

func MergeActionEvalKey(targetVar, key string) string

MergeActionEvalKey composes the map key under which a MERGE ON CREATE / ON MATCH property-set action's per-row RHS evaluator is registered and looked up. targetVar is the entity variable the action writes (a node, a bound endpoint, or a relationship variable) and key is the property key. The NUL separator cannot appear in a Cypher identifier, so the composed key is unambiguous across distinct (variable, property) pairs. The physical builder ([cypher] package) and the operators here must agree on this encoding, so it is defined once and exported.

func NewUniqueBackingIndex added in v0.2.0 

func NewUniqueBackingIndex() index.Subscriber

NewUniqueBackingIndex returns a fresh unbound hash-index subscriber. It is kept for callers (tests, recovery) that do not have a live graph and therefore cannot build a bound index. The engine's write path always supplies a bound subscriber via CreateConstraintOp.WithBackingIndex so the index self-maintains from the change fan-out.

func PropMapContainsNullLiteral 

func PropMapContainsNullLiteral(s string) bool

PropMapContainsNullLiteral reports whether the property-map source string contains any value that is the literal `null`. It is used by MERGE to surface the openCypher `MergeReadOwnWrites` error when a merge predicate contains a null property value — such a merge can never match its own write because null comparisons are always tri-valued false, so the engine rejects the pattern outright. Used at MERGE plan-build time; CREATE silently drops null-valued properties so it does NOT use this check.

The argument may be either a single map literal "{k: v, …}" or a larger surface form (e.g. a full pattern string "(a)-[r:T {k: null}]->(b)") in which case every embedded balanced "{...}" segment is scanned. The check splits each map at top-level commas, isolates each value substring, and reports true if any value (case-insensitively) equals the bare token `null`. Variable refs and expressions are ignored — they parse to non-literal forms.

func UniqueIndexName added in v0.2.0 

func UniqueIndexName(label, prop string) string

UniqueIndexName returns the deterministic synthetic name of the hash index that backs a UNIQUE constraint on (label, prop). It is exported so the engine can re-create the same backing index when re-registering a constraint recovered from disk, keeping the name in lockstep with the one the CreateConstraint operator uses.

func WithCyphermorphism 

func WithCyphermorphism(relCols []int) expandOption

WithCyphermorphism returns an option that enables cyphermorphism enforcement. relCols lists the column indices in each input row that hold existing edge IDs (as expr.IntegerValue). When Expand is about to emit a row with a new edgeID, it rejects the row if edgeID equals any value already present in those columns.

Types

type AllNodesScan 

type AllNodesScan struct {
	// contains filtered or unexported fields
}

AllNodesScan is a Volcano leaf operator that produces one Row per node in the graph. Each Row has a single column: an expr.IntegerValue holding the node's graph.NodeID cast to int64.

AllNodesScan is NOT safe for concurrent use.

func NewAllNodesScan 

func NewAllNodesScan(g nodeWalker) *AllNodesScan

NewAllNodesScan creates an AllNodesScan over g.

func (*AllNodesScan) Close 

func (op *AllNodesScan) Close() error

Close releases resources. The collected nodeIDs slice is retained (but its length zeroed) to allow reuse if Init is called again.

func (*AllNodesScan) Init 

func (op *AllNodesScan) Init(ctx context.Context) error

Init collects all NodeIDs from the graph into an internal slice. The collection itself honours ctx cancellation every 4096 nodes.

func (*AllNodesScan) Next 

func (op *AllNodesScan) Next(out *Row) (bool, error)

Next writes the next NodeID into out and returns (true, nil), or returns (false, nil) at end-of-stream. ctx.Err() is checked on every call.

type AllShortestPaths 

type AllShortestPaths struct {
	// contains filtered or unexported fields
}

AllShortestPaths is a Volcano pipeline operator that, for each input row, finds all paths of minimum length from srcCol to dstCol and emits one output row per path, each carrying the flat alternating path list (stride VLEHopStride).

AllShortestPaths is NOT safe for concurrent use.

func NewAllShortestPaths 

func NewAllShortestPaths(input Operator, fwd, rev csrAdjacency, dir Direction, srcCol, dstCol int) *AllShortestPaths

NewAllShortestPaths creates an AllShortestPaths operator. Like NewShortestPath it starts with no type filter and minHops == 1; configure via AllShortestPaths.WithTypeFilter and AllShortestPaths.WithHopBounds.

func (*AllShortestPaths) Close 

func (op *AllShortestPaths) Close() error

Close closes the input operator.

func (*AllShortestPaths) Init 

func (op *AllShortestPaths) Init(ctx context.Context) error

Init initialises the operator.

func (*AllShortestPaths) Next 

func (op *AllShortestPaths) Next(out *Row) (bool, error)

Next emits one row per shortest path per input row. With no path: one Null-path row (OPTIONAL MATCH) or no row (MATCH).

func (*AllShortestPaths) WithHopBounds added in v0.6.0 

func (op *AllShortestPaths) WithHopBounds(minHops, maxHops int) *AllShortestPaths

WithHopBounds sets the accepted path-length window. It returns op for chaining.

func (*AllShortestPaths) WithOptional added in v0.6.0 

func (op *AllShortestPaths) WithOptional(optional bool) *AllShortestPaths

WithOptional selects the OPTIONAL MATCH no-path behaviour (emit a single Null-path row) when optional is true; the default (false) drops the row. It returns op for chaining.

func (*AllShortestPaths) WithPathPredicate added in v0.6.0 

func (op *AllShortestPaths) WithPathPredicate(pred func(Row) (bool, error)) *AllShortestPaths

WithPathPredicate fuses a whole-path predicate onto the operator (#1786). The operator then returns ALL shortest paths that SATISFY pred (an exhaustive search at the minimum satisfying length), instead of all unconstrained shortest paths. pred is called with each candidate's full output row. It returns op for chaining.

func (*AllShortestPaths) WithTypeFilter added in v0.6.0 

func (op *AllShortestPaths) WithTypeFilter(edgeType string, filter map[uint64]string) *AllShortestPaths

WithTypeFilter restricts traversal to edges whose forward position is present in filter. It returns op for chaining.

func (*AllShortestPaths) WithWorkBudget added in v0.7.0 

func (op *AllShortestPaths) WithWorkBudget(maxPerRow, maxTotal int) *AllShortestPaths

WithWorkBudget overrides the exhaustive path-predicate search's per-input-row and aggregate per-query edge-traversal caps (see ShortestPath.WithWorkBudget). A non-positive value leaves the corresponding default in place. It returns op for chaining and is primarily a testing seam; production uses the defaults.

type AntiSemiApply 

type AntiSemiApply struct {
	// contains filtered or unexported fields
}

AntiSemiApply emits each outer row for which the inner sub-plan produces zero rows.

AntiSemiApply is NOT safe for concurrent use.

func NewAntiSemiApply 

func NewAntiSemiApply(outer, inner Operator, arg *Argument) *AntiSemiApply

NewAntiSemiApply creates an AntiSemiApply operator.

  • outer is the driving (left) plan.
  • inner is the correlated (right) sub-plan whose leaf is arg.
  • arg is the Argument node seeded with each outer row before inner Init.

func (*AntiSemiApply) Close 

func (op *AntiSemiApply) Close() error

Close closes the outer plan.

func (*AntiSemiApply) Init 

func (op *AntiSemiApply) Init(ctx context.Context) error

Init initialises the outer plan.

func (*AntiSemiApply) Next 

func (op *AntiSemiApply) Next(out *Row) (bool, error)

Next advances to the next outer row for which the inner plan has zero results.

type Apply 

type Apply struct {
	// contains filtered or unexported fields
}

Apply is a Volcano pipeline operator that performs a dependent (correlated) join: for each outer row, it re-runs the inner plan seeded with that row and emits one output row per inner result.

Apply is NOT safe for concurrent use.

func NewApply 

func NewApply(outer, inner Operator, arg *Argument) *Apply

NewApply creates an Apply operator.

  • outer is the left (driving) plan.
  • inner is the right (sub) plan; its leaf must be the provided arg.
  • arg is the Argument node at the root of inner; Apply seeds it before each inner Init call.

Apply takes ownership of both plans. The caller must not use outer or inner directly after calling NewApply.

func (*Apply) Close 

func (op *Apply) Close() error

Close releases resources and closes both the outer and inner plans.

func (*Apply) Init 

func (op *Apply) Init(ctx context.Context) error

Init initialises both the outer plan and stores ctx for subsequent Next calls. The inner plan is initialised lazily on the first outer row.

func (*Apply) Next 

func (op *Apply) Next(out *Row) (bool, error)

Next advances the Apply operator:

  • If there is no current inner row available, it pulls the next outer row, seeds and re-inits the inner plan, then pulls from the inner plan.
  • Returns each combined (outer || inner) row.
  • Returns (false, nil) when the outer plan is exhausted.

type Argument 

type Argument struct {
	// contains filtered or unexported fields
}

Argument is a Volcano leaf operator that emits the single outer Row injected by an Apply driver. It emits exactly one row per Init/Next cycle.

Argument is NOT safe for concurrent use.

func NewArgument 

func NewArgument() *Argument

NewArgument creates an Argument operator with no initial row. The Apply driver must call [SetOuterRow] before the first Init/Next cycle.

func (*Argument) Close 

func (op *Argument) Close() error

Close is a no-op; Argument holds no resources beyond the outer row reference.

func (*Argument) Init 

func (op *Argument) Init(ctx context.Context) error

Init resets the emitted flag so that the next Next call returns the outer row. It does not require a child operator.

func (*Argument) Next 

func (op *Argument) Next(out *Row) (bool, error)

Next emits the outer row exactly once per Init call. Subsequent calls return (false, nil) until Init is called again.

func (*Argument) SetOuterRow 

func (op *Argument) SetOuterRow(row Row)

SetOuterRow injects the current outer row. It must be called by the Apply driver before each Init/Next cycle for the inner plan.

type ConstraintInfo added in v0.2.0 

type ConstraintInfo struct {
	// KindUnique is true for a UNIQUE constraint, false for NOT NULL.
	KindUnique bool
	// Label is the constrained node label.
	Label string
	// Property is the constrained property key.
	Property string
	// Name is the user-defined constraint name (may be empty for a constraint
	// registered without one).
	Name string
}

ConstraintInfo is a structured description of one registered constraint, used to persist the constraint set durably and to re-register it on recovery. KindUnique distinguishes UNIQUE (true) from NOT NULL (false).

type ConstraintKind 

type ConstraintKind uint8

ConstraintKind distinguishes UNIQUE from NOT_NULL constraints. 

const (
	// ConstraintUnique requires that at most one node with a given label has a
	// particular value for the constrained property.
	ConstraintUnique ConstraintKind = iota
	// ConstraintNotNull requires that every node with a given label has a
	// non-null value for the constrained property.
	ConstraintNotNull
)

type ConstraintRegistry 

type ConstraintRegistry struct {
	// contains filtered or unexported fields
}

ConstraintRegistry is a thread-safe registry of active constraints. It stores unique and not-null constraints keyed by "label.prop".

ConstraintRegistry is safe for concurrent use.

func NewConstraintRegistry 

func NewConstraintRegistry() *ConstraintRegistry

NewConstraintRegistry creates an empty ConstraintRegistry.

func (*ConstraintRegistry) CheckSetProperty 

func (r *ConstraintRegistry) CheckSetProperty(labels []string, prop string, value lpg.PropertyValue, mgr *index.Manager) error

CheckSetProperty validates that setting prop = value on a node with the given labels does not violate any registered constraint. mgr is used for unique-constraint index lookups (hash index Cardinality check) as a secondary source; the primary source is the registry's own value set.

The primary value-set is authoritative: when it is present (non-nil) and does not contain value, the secondary hash-index check is skipped. The value-set is seeded from the live graph at constraint creation and kept current by RecordPropertySet / ReleasePropertyValue / ReseedFromGraph, so an absent primary "not present" signal means the value is genuinely free. The secondary check is consulted only when the primary set has not been initialised (nil), covering the narrow window before SeedUniqueValues is called.

Returns *ConstraintViolationError (which wraps ErrConstraintViolation) on the first violation found; nil when all constraints pass.

func (*ConstraintRegistry) Constraints added in v0.2.0 

func (r *ConstraintRegistry) Constraints() []ConstraintInfo

Constraints returns a structured snapshot of every registered constraint, in deterministic order (UNIQUE before NOT NULL, then by label, property, name). It is used to persist the constraint set into a snapshot and to compare the recovered set against the live one.

Constraints is safe for concurrent use.

func (*ConstraintRegistry) Count added in v0.3.0 

func (r *ConstraintRegistry) Count() int

Count returns the number of registered constraints (UNIQUE + NOT NULL). It is a cheap, allocation-free alternative to len(Constraints()) that the engine uses to mirror the count onto the graph for the checkpointer (#1464).

Count is safe for concurrent use.

func (*ConstraintRegistry) HasAnyNotNull added in v0.6.0 

func (r *ConstraintRegistry) HasAnyNotNull() bool

HasAnyNotNull reports whether at least one NOT NULL (property-existence) constraint is registered. It is the cheap gate the engine consults at every commit before doing any touched-node existence-constraint work: when it returns false the commit-time check is a no-op and the per-transaction touched-node recording is skipped entirely, so a workload with no existence constraints pays only this single lock-protected length read per commit.

HasAnyNotNull is safe for concurrent use.

func (*ConstraintRegistry) HasNotNull 

func (r *ConstraintRegistry) HasNotNull(label, prop string) bool

HasNotNull reports whether a not-null constraint exists for (label, prop).

func (*ConstraintRegistry) HasUnique added in v0.2.0 

func (r *ConstraintRegistry) HasUnique(label, prop string) bool

HasUnique reports whether a unique constraint exists for (label, prop).

func (*ConstraintRegistry) ListConstraintRows 

func (r *ConstraintRegistry) ListConstraintRows() [][]expr.Value

ListConstraintRows returns a [][]expr.Value where each inner slice has four elements: [name, type, label, property]. The name column carries the constraint's declared (or auto-generated) name — the same name DROP CONSTRAINT resolves by — falling back to the canonical "label.prop" key only for a constraint registered without a name (the legacy anonymous path). type is "UNIQUE" or "NOT_NULL". Rows are returned in deterministic order (name, type, label, property).

ListConstraintRows is safe for concurrent use.

func (*ConstraintRegistry) NameInUse added in v0.8.0 

func (r *ConstraintRegistry) NameInUse(name string) (kind ConstraintKind, label, prop string, found bool)

NameInUse reports the identity of the constraint currently holding name, or found=false when no constraint carries it. It is the CREATE-time lookup used to reject a name already used by a different constraint. NameInUse is safe for concurrent use and deterministic (see [ResolveByName]).

func (*ConstraintRegistry) NotNullProperties added in v0.6.0 

func (r *ConstraintRegistry) NotNullProperties(label string) []string

NotNullProperties returns the property keys for which a NOT NULL constraint is registered on label, or nil when label carries no existence constraint. It is the per-label lookup the commit-time existence check uses to test only the constrained properties of a touched node's labels.

The lookup is O(1) via the notNullByLabel index and allocates nothing on this hot path: it returns the registry's own copy-on-write slice, which RegisterNotNull / UnregisterNotNull never mutate in place. The caller must treat the result as READ-ONLY (it is shared and must not be appended to or modified). Most labels carry zero or one existence constraint, so the common return is nil or a one-element slice. NotNullProperties is safe for concurrent use.

func (*ConstraintRegistry) RecordPropertySet 

func (r *ConstraintRegistry) RecordPropertySet(labels []string, prop string, value lpg.PropertyValue)

RecordPropertySet records that a property value has been successfully written to a node with the given labels. This keeps the unique value sets up-to-date so that subsequent CheckSetProperty calls detect violations. It is a no-op when no unique constraint exists for (label, prop).

func (*ConstraintRegistry) RegisterNotNull 

func (r *ConstraintRegistry) RegisterNotNull(label, prop string)

RegisterNotNull adds a not-null constraint for (label, prop).

func (*ConstraintRegistry) RegisterUnique 

func (r *ConstraintRegistry) RegisterUnique(label, prop, indexName string)

RegisterUnique adds a unique constraint for (label, prop) backed by indexName in the index.Manager.

func (*ConstraintRegistry) ReleasePropertyValue added in v0.3.0 

func (r *ConstraintRegistry) ReleasePropertyValue(labels []string, prop string, value lpg.PropertyValue)

ReleasePropertyValue removes a value from the unique value-set so it is no longer treated as "in use" by the constraint. It must be called whenever a previously recorded value is no longer present in the graph: on node deletion, on REMOVE of a constrained property, and when SET replaces an existing constrained value with a new one (releasing the old value).

ReleasePropertyValue is a no-op when the value was not present in the value-set, or when no unique constraint exists for (label, prop). It is safe for concurrent use.

func (*ConstraintRegistry) ReseedFromGraph added in v0.3.0 

func (r *ConstraintRegistry) ReseedFromGraph(scanFn func(label, prop string) []lpg.PropertyValue)

ReseedFromGraph clears every UNIQUE value-set and rebuilds it from the provided (label, prop) → values mapping. It is called after an in-memory transaction undo (rollback) so the registry reflects the restored graph state rather than the rolled-back writes.

The caller supplies a scanFn that, given a label and property key, returns the property values currently carried by all live nodes with that label. ReseedFromGraph acquires the registry write lock only once per constraint, so the scan itself must NOT hold the registry lock.

func (*ConstraintRegistry) ResolveByName added in v0.4.0 

func (r *ConstraintRegistry) ResolveByName(name string) (kind ConstraintKind, label, prop string, found bool)

ResolveByName resolves a user-defined constraint name to its (kind, label, property) identity, so a DROP CONSTRAINT <name> can locate the constraint to remove. It searches the UNIQUE names first, then the NOT NULL names, and returns found=false when no constraint carries that name.

Only constraints registered WITH a name (via [SetConstraintName], which the CREATE CONSTRAINT executor calls) are resolvable; an anonymous constraint registered through the legacy [RegisterUnique] / [RegisterNotNull] path has no name to match.

ResolveByName is safe for concurrent use. It is deterministic: it scans the UNIQUE names then the NOT NULL names, each in sorted key order, and returns the first match, so a lookup never depends on Go map iteration order. Once constraint names are enforced unique at CREATE time (ErrConstraintNameConflict) at most one constraint can match, so the sorted scan is belt-and-braces.

func (*ConstraintRegistry) SeedUniqueValues added in v0.2.0 

func (r *ConstraintRegistry) SeedUniqueValues(label, prop string, values []lpg.PropertyValue) error

SeedUniqueValues populates the value-set of an already-registered UNIQUE constraint on (label, prop) from the property values of the nodes that currently carry the label. It is the post-creation seed that makes a constraint added to a non-empty dataset functional: without it the value-set starts empty and pre-existing duplicates (or duplicates of a pre-existing value) are accepted on the next write.

It also enforces the at-creation invariant (Neo4j semantics, audit gap H2): if two of the supplied values are equal it returns a *ConstraintViolationError wrapping ErrConstraintViolation and seeds nothing, so the caller can reject CREATE CONSTRAINT over already-duplicated data. Null values (the zero PropertyValue) are ignored by a UNIQUE constraint and are skipped.

SeedUniqueValues is a no-op (and returns nil) when no UNIQUE constraint is registered for (label, prop).

func (*ConstraintRegistry) SeedUniqueValuesIgnoringDuplicates added in v0.2.0 

func (r *ConstraintRegistry) SeedUniqueValuesIgnoringDuplicates(label, prop string, values []lpg.PropertyValue)

SeedUniqueValuesIgnoringDuplicates seeds the value-set of an already-registered UNIQUE constraint on (label, prop) from values WITHOUT rejecting pre-existing duplicates. It is the recovery seed: recovery must always succeed so the store is serviceable, and a duplicate that predates the constraint is a historical artefact the live enforcement path still rejects on the next write. Null values are skipped. No-op when no UNIQUE constraint is registered for (label, prop).

func (*ConstraintRegistry) SetConstraintName added in v0.2.0 

func (r *ConstraintRegistry) SetConstraintName(kindUnique bool, label, prop, name string)

SetConstraintName records the user-defined name of the constraint of the given kind on (label, prop), so the constraint round-trips durably with the name the client declared. kindUnique selects UNIQUE (true) vs NOT NULL. A later [UnregisterUnique] / [UnregisterNotNull] clears the matching name.

func (*ConstraintRegistry) UniqueIndexName 

func (r *ConstraintRegistry) UniqueIndexName(label, prop string) (string, bool)

UniqueIndexName returns the backing index name for a unique constraint on (label, prop), or ("", false) if none exists.

func (*ConstraintRegistry) UnregisterNotNull 

func (r *ConstraintRegistry) UnregisterNotNull(label, prop string)

UnregisterNotNull removes the not-null constraint for (label, prop). No-op if absent.

func (*ConstraintRegistry) UnregisterUnique 

func (r *ConstraintRegistry) UnregisterUnique(label, prop string)

UnregisterUnique removes the unique constraint for (label, prop). No-op if absent.

type ConstraintViolationError 

type ConstraintViolationError struct {
	// Label is the node label the constraint is defined on.
	Label string
	// Property is the constrained property key.
	Property string
	// Kind describes the type of constraint: "UNIQUE" or "NOT NULL".
	Kind string
	// Detail is an optional human-readable explanation.
	Detail string
}

ConstraintViolationError carries structured context about which constraint was violated.

func (*ConstraintViolationError) Error 

func (e *ConstraintViolationError) Error() string

Error implements the error interface.

func (*ConstraintViolationError) Unwrap 

func (e *ConstraintViolationError) Unwrap() error

Unwrap chains to ErrConstraintViolation so callers can use errors.Is.

type CorrelatedApply 

type CorrelatedApply struct {
	// contains filtered or unexported fields
}

CorrelatedApply is a Volcano pipeline operator that performs a dependent (correlated) join with the convention that the inner pipeline begins with an Argument leaf re-emitting the outer row. The inner row is forwarded verbatim as the operator's output; no concatenation with the outer row is performed (the outer columns are already present in the inner row).

CorrelatedApply is NOT safe for concurrent use.

func NewCorrelatedApply 

func NewCorrelatedApply(outer, inner Operator, arg *Argument) *CorrelatedApply

NewCorrelatedApply creates a CorrelatedApply operator.

  • outer is the left (driving) plan.
  • inner is the right (sub) plan whose leftmost leaf is the provided arg.
  • arg is the Argument node at the inner leaf; CorrelatedApply seeds it before each inner Init call so that the inner pipeline observes the current outer row.

CorrelatedApply takes ownership of both plans. The caller must not use outer or inner directly after calling NewCorrelatedApply.

func (*CorrelatedApply) Close 

func (op *CorrelatedApply) Close() error

Close releases resources and closes both the outer and inner plans.

func (*CorrelatedApply) Init 

func (op *CorrelatedApply) Init(ctx context.Context) error

Init initialises the outer plan and stores ctx for subsequent Next calls. The inner plan is initialised lazily on the first outer row.

func (*CorrelatedApply) Next 

func (op *CorrelatedApply) Next(out *Row) (bool, error)

Next advances the CorrelatedApply operator. The inner row is forwarded verbatim; CorrelatedApply does not concatenate the outer row because the inner row already carries the outer columns (the inner pipeline's leaf Argument re-emitted them).

type CreateConstraintOp 

type CreateConstraintOp struct {
	// contains filtered or unexported fields
}

CreateConstraintOp is a Volcano DDL operator that registers a constraint.

CreateConstraintOp is NOT safe for concurrent use.

func NewCreateConstraintOp 

func NewCreateConstraintOp(
	name, label, prop string,
	kind ConstraintKind,
	ifNotExists bool,
	mgr *index.Manager,
	reg *ConstraintRegistry,
	onSchemaChange func(),
) *CreateConstraintOp

NewCreateConstraintOp creates a CreateConstraintOp. onSchemaChange, when non-nil, is invoked exactly once after the operator successfully registers a new constraint — i.e. NOT when the IF NOT EXISTS branch silently absorbs an already-registered constraint. The Engine wires e.ClearPlanCache as onSchemaChange so cached plans are invalidated after a real schema mutation.

func (*CreateConstraintOp) Close 

func (op *CreateConstraintOp) Close() error

Close implements Operator.

func (*CreateConstraintOp) Init 

func (op *CreateConstraintOp) Init(ctx context.Context) error

Init implements Operator.

func (*CreateConstraintOp) Next 

func (op *CreateConstraintOp) Next(_ *Row) (bool, error)

Next implements Operator. Performs the CREATE CONSTRAINT side effect on the first call, then signals end-of-stream.

func (*CreateConstraintOp) WithBackingIndex added in v0.3.0 

func (op *CreateConstraintOp) WithBackingIndex(sub index.Subscriber) *CreateConstraintOp

WithBackingIndex supplies a pre-built index subscriber to use as the UNIQUE constraint's backing hash index instead of creating a fresh unbound index. The engine passes a bound index (built and backfilled from the live graph) so the index self-maintains from the change fan-out at commit time. Returns op for chaining.

type CreateIndexOp 

type CreateIndexOp struct {
	// contains filtered or unexported fields
}

CreateIndexOp is a Volcano DDL operator that registers a new secondary index.

CreateIndexOp is NOT safe for concurrent use.

func NewCreateIndexOp 

func NewCreateIndexOp(
	name string,
	kind IndexKindExec,
	ifNotExists bool,
	mgr *index.Manager,
	onSchemaChange func(),
) *CreateIndexOp

NewCreateIndexOp creates a CreateIndexOp. onSchemaChange, when non-nil, is invoked exactly once after the operator successfully creates a new index in mgr — i.e. NOT when the IF NOT EXISTS branch silently absorbs a duplicate. The Engine wires e.ClearPlanCache as onSchemaChange so cached plans are invalidated after a real schema mutation.

func (*CreateIndexOp) Close 

func (op *CreateIndexOp) Close() error

Close implements Operator.

func (*CreateIndexOp) Init 

func (op *CreateIndexOp) Init(ctx context.Context) error

Init implements Operator.

func (*CreateIndexOp) Next 

func (op *CreateIndexOp) Next(_ *Row) (bool, error)

Next implements Operator. It performs the CREATE INDEX side effect on the first call, then signals end-of-stream. Returns (false, nil) immediately on subsequent calls.

type CreateNode 

type CreateNode struct {
	// contains filtered or unexported fields
}

CreateNode creates a new graph node per input row, sets its labels and properties, and appends the new NodeID as a new column.

CreateNode is NOT safe for concurrent use.

func NewCreateNode 

func NewCreateNode(
	nodeVar string,
	labels []string,
	properties string,
	child Operator,
	mutator GraphMutator,
) (*CreateNode, error)

NewCreateNode creates a CreateNode operator.

nodeVar is the variable name bound to the new node (may be empty if the node is not referenced downstream). labels is the ordered list of labels to attach. properties is the opaque literal property-map string (e.g. `{name: "Alice"}`) produced by the IR translator; it is parsed once during construction. mutator is the graph write surface.

func (*CreateNode) Close 

func (op *CreateNode) Close() error

Close closes the child operator.

func (*CreateNode) Init 

func (op *CreateNode) Init(ctx context.Context) error

Init initialises the operator and its child.

The first CreateNode.Init in the process also seeds [globalNodeCounter] past the largest synthetic key currently interned in op.mutator, so that node keys generated in this process cannot collide with keys persisted by an earlier process and replayed during WAL / snapshot recovery. The seed is gated by [globalNodeCounterSeededOnce] so the scan runs at most once per process regardless of how many CreateNode operators are created.

func (*CreateNode) Next 

func (op *CreateNode) Next(out *Row) (bool, error)

Next pulls one row from the child, creates a node, and appends the NodeID column. Returns (true, nil) when a row was produced, (false, nil) at end-of-stream, (false, err) on error.

func (*CreateNode) WithConstraints 

func (op *CreateNode) WithConstraints(reg *ConstraintRegistry, mgr *index.Manager) *CreateNode

WithConstraints attaches a ConstraintRegistry and index.Manager to the operator for pre-write enforcement. Both must be non-nil. Returns op for chaining.

func (*CreateNode) WithParams 

func (op *CreateNode) WithParams(params map[string]expr.Value) (*CreateNode, error)

WithParams attaches query parameters for $name substitution in property expressions. Re-parses the property map with the supplied params. Returns op for chaining.

func (*CreateNode) WithPropsEvalFn 

func (op *CreateNode) WithPropsEvalFn(fn PropsEvalFn) *CreateNode

WithPropsEvalFn attaches a per-row property evaluator. When fn is non-nil it is called on every Next invocation and its results are merged with the statically parsed props (literal values). Dynamic results take precedence over same-keyed literal values, allowing the property map to contain a mix of literals and expression-valued entries.

type CreateRelationship 

type CreateRelationship struct {
	// contains filtered or unexported fields
}

CreateRelationship creates a new directed edge per input row between two already-bound nodes.

CreateRelationship is NOT safe for concurrent use.

func NewCreateRelationship 

func NewCreateRelationship(
	startVar, endVar, relVar, relType, properties string,
	schema map[string]int,
	child Operator,
	mutator GraphMutator,
) (*CreateRelationship, error)

NewCreateRelationship creates a CreateRelationship operator.

startVar and endVar are the variable names (column indices are looked up in schema) of the source and destination nodes. relVar is the variable name bound to the new relationship (may be empty). relType is the relationship type label. properties is the opaque literal property-map string. schema maps currently bound variable names to their column indices.

func (*CreateRelationship) Close 

func (op *CreateRelationship) Close() error

Close closes the child operator.

func (*CreateRelationship) Init 

func (op *CreateRelationship) Init(ctx context.Context) error

Init initialises the operator and its child.

func (*CreateRelationship) Next 

func (op *CreateRelationship) Next(out *Row) (bool, error)

Next pulls one row from the child, resolves the endpoint NodeIDs, creates the edge, and appends an optional RelationshipValue column.

func (*CreateRelationship) WithParams 

func (op *CreateRelationship) WithParams(params map[string]expr.Value) (*CreateRelationship, error)

WithParams re-parses the property map with the supplied query parameters for $name substitution. Returns op for chaining.

func (*CreateRelationship) WithPropsEvalFn 

func (op *CreateRelationship) WithPropsEvalFn(fn PropsEvalFn) *CreateRelationship

WithPropsEvalFn attaches a per-row property evaluator. See CreateNode.WithPropsEvalFn.

type DeleteNode 

type DeleteNode struct {
	// contains filtered or unexported fields
}

DeleteNode deletes an already-bound node (labels + properties stripped) from the graph, provided it has no incident relationships.

DeleteNode is NOT safe for concurrent use.

func NewDeleteNode 

func NewDeleteNode(
	nodeVar string,
	schema map[string]int,
	child Operator,
	mutator GraphMutator,
) *DeleteNode

NewDeleteNode creates a DeleteNode operator.

func (*DeleteNode) Close 

func (op *DeleteNode) Close() error

Close closes the child operator.

func (*DeleteNode) Init 

func (op *DeleteNode) Init(ctx context.Context) error

Init initialises the operator and its child.

func (*DeleteNode) Next 

func (op *DeleteNode) Next(out *Row) (bool, error)

Next pulls one row from the child and deletes the bound node.

func (*DeleteNode) WithConstraintRegistry added in v0.3.0 

func (op *DeleteNode) WithConstraintRegistry(reg *ConstraintRegistry) *DeleteNode

WithConstraintRegistry attaches a ConstraintRegistry so DeleteNode releases unique-constraint value reservations when a node is deleted. Returns op for chaining.

func (*DeleteNode) WithRelEndpoints 

func (op *DeleteNode) WithRelEndpoints(fn RelEndpointFn) *DeleteNode

WithRelEndpoints attaches a per-row lookup that returns the (srcID, dstID) endpoints of the edge identified by the bare-variable target. When set AND the schema-direct slot holds an IntegerValue (the in-pipeline edge-id encoding emitted by Expand), the operator dispatches to the edge-removal path instead of treating the integer as a NodeID.

func (*DeleteNode) WithTargetEvalFn 

func (op *DeleteNode) WithTargetEvalFn(fn TargetEvalFn) *DeleteNode

WithTargetEvalFn attaches a per-row evaluator for non-variable DELETE targets (subscripts, property access, …). When set, the operator resolves the target value via the evaluator instead of the schema lookup keyed by nodeVar.

type DeleteRelationship 

type DeleteRelationship struct {
	// contains filtered or unexported fields
}

DeleteRelationship removes a directed edge per input row.

DeleteRelationship is NOT safe for concurrent use.

func NewDeleteRelationship 

func NewDeleteRelationship(
	relVar string,
	schema map[string]int,
	child Operator,
	mutator GraphMutator,
) *DeleteRelationship

NewDeleteRelationship creates a DeleteRelationship operator.

func (*DeleteRelationship) Close 

func (op *DeleteRelationship) Close() error

Close closes the child operator.

func (*DeleteRelationship) Init 

func (op *DeleteRelationship) Init(ctx context.Context) error

Init initialises the operator and its child.

func (*DeleteRelationship) Next 

func (op *DeleteRelationship) Next(out *Row) (bool, error)

Next pulls one row from the child and removes the bound relationship.

type DetachDelete 

type DetachDelete struct {
	// contains filtered or unexported fields
}

DetachDelete removes all incident edges from a node and then strips the node's labels and properties.

DetachDelete is NOT safe for concurrent use.

func NewDetachDelete 

func NewDetachDelete(
	nodeVar string,
	schema map[string]int,
	child Operator,
	mutator GraphMutator,
) *DetachDelete

NewDetachDelete creates a DetachDelete operator.

func (*DetachDelete) Close 

func (op *DetachDelete) Close() error

Close closes the child operator.

func (*DetachDelete) Init 

func (op *DetachDelete) Init(ctx context.Context) error

Init initialises the operator and its child.

func (*DetachDelete) Next 

func (op *DetachDelete) Next(out *Row) (bool, error)

Next pulls one row from the child, removes all incident edges of the bound node, then strips the node's labels and properties.

func (*DetachDelete) WithConstraintRegistry added in v0.3.0 

func (op *DetachDelete) WithConstraintRegistry(reg *ConstraintRegistry) *DetachDelete

WithConstraintRegistry attaches a ConstraintRegistry so DetachDelete releases unique-constraint value reservations when a node is deleted. Returns op for chaining.

func (*DetachDelete) WithTargetEvalFn 

func (op *DetachDelete) WithTargetEvalFn(fn TargetEvalFn) *DetachDelete

WithTargetEvalFn attaches a per-row evaluator for non-variable DETACH DELETE targets (subscripts, property access, …).

type Direction 

type Direction uint8

Direction controls which edges Expand follows. 

const (
	// DirOut follows only out-edges of the source node.
	DirOut Direction = iota + 1
	// DirIn follows only in-edges (reverse edges).
	DirIn
	// DirBoth follows both out-edges and in-edges.
	DirBoth
)

type Distinct 

type Distinct struct {
	// contains filtered or unexported fields
}

Distinct is a streaming Volcano operator that emits each unique row exactly once. It maintains a hash set of seen rows; hash collisions are resolved by full equality checks.

Distinct is NOT safe for concurrent use.

func NewDistinct 

func NewDistinct(child Operator, maxDistinct int) *Distinct

NewDistinct creates a Distinct operator.

  • child: the upstream operator.
  • maxDistinct: upper bound on distinct rows; pass 0 to use DefaultMaxDistinct.

func (*Distinct) Close 

func (op *Distinct) Close() error

Close closes the child operator and releases internal state.

func (*Distinct) Init 

func (op *Distinct) Init(ctx context.Context) error

Init initialises the operator and resets the deduplication state.

func (*Distinct) Next 

func (op *Distinct) Next(out *Row) (bool, error)

Next pulls rows from the child and emits the first occurrence of each unique row. Duplicate rows are silently discarded. Returns ErrDistinctMemoryExceeded if more than maxDistinct distinct rows are encountered.

func (*Distinct) WithByteBudget added in v0.7.0 

func (op *Distinct) WithByteBudget(maxBytes int64, estimateRow func(Row) int64) *Distinct

WithByteBudget bounds the estimated retained size of the stored distinct rows by maxBytes. It complements the maxDistinct count cap so a few large-valued distinct rows cannot exceed the engine's result-byte budget before the count cap fires (#1841). A non-positive maxBytes or nil estimateRow leaves the byte dimension disabled. Returns op for chaining and must be called before Init.

type DropConstraintOp 

type DropConstraintOp struct {
	// contains filtered or unexported fields
}

DropConstraintOp is a Volcano DDL operator that deregisters a constraint.

DropConstraintOp is NOT safe for concurrent use.

func NewDropConstraintOp 

func NewDropConstraintOp(
	name, label, prop string,
	kind ConstraintKind,
	ifExists bool,
	mgr *index.Manager,
	reg *ConstraintRegistry,
	onSchemaChange func(),
) *DropConstraintOp

NewDropConstraintOp creates a DropConstraintOp. onSchemaChange, when non-nil, is invoked exactly once after the operator successfully removes a constraint — i.e. NOT when the IF EXISTS branch silently absorbs an absent-constraint condition. The Engine wires e.ClearPlanCache as onSchemaChange so cached plans are invalidated after a real schema mutation.

func (*DropConstraintOp) Close 

func (op *DropConstraintOp) Close() error

Close implements Operator.

func (*DropConstraintOp) Init 

func (op *DropConstraintOp) Init(ctx context.Context) error

Init implements Operator.

func (*DropConstraintOp) Next 

func (op *DropConstraintOp) Next(_ *Row) (bool, error)

Next implements Operator. Performs the DROP CONSTRAINT side effect on the first call, then signals end-of-stream.

type DropIndexOp 

type DropIndexOp struct {
	// contains filtered or unexported fields
}

DropIndexOp is a Volcano DDL operator that deregisters a secondary index.

DropIndexOp is NOT safe for concurrent use.

func NewDropIndexOp 

func NewDropIndexOp(
	name string,
	ifExists bool,
	mgr *index.Manager,
	onSchemaChange func(),
) *DropIndexOp

NewDropIndexOp creates a DropIndexOp. onSchemaChange, when non-nil, is invoked exactly once after the operator successfully drops the index — i.e. NOT when the IF EXISTS branch silently absorbs a missing-index error. The Engine wires e.ClearPlanCache as onSchemaChange so cached plans are invalidated after a real schema mutation.

func (*DropIndexOp) Close 

func (op *DropIndexOp) Close() error

Close implements Operator.

func (*DropIndexOp) Init 

func (op *DropIndexOp) Init(ctx context.Context) error

Init implements Operator.

func (*DropIndexOp) Next 

func (op *DropIndexOp) Next(_ *Row) (bool, error)

Next implements Operator. It performs the DROP INDEX side effect on the first call, then signals end-of-stream.

type Eager 

type Eager struct {
	// contains filtered or unexported fields
}

Eager is a pipeline-breaking barrier. Init drains every row from the child into an internal buffer; Next then re-emits the buffered rows in insertion order.

Eager is NOT safe for concurrent use: it holds unsynchronised buffer and cursor state mutated across Init/Next/Close, so a single goroutine must drive one operator tree, like every other Operator.

func NewEager 

func NewEager(child Operator, maxRows int) *Eager

NewEager wraps child in an Eager barrier.

  • child: the upstream operator to drain.
  • maxRows: upper bound on rows held in memory; pass 0 to use DefaultMaxEagerRows.

func (*Eager) Close 

func (op *Eager) Close() error

Close releases the buffer and closes the child.

func (*Eager) Init 

func (op *Eager) Init(ctx context.Context) error

Init initialises the operator AND drains every child row into the buffer so the downstream pipeline can apply LIMIT / SKIP / DISTINCT without starving any write-side child of its driving rows.

The drain is bounded by maxRows (ErrEagerMemoryExceeded when exceeded) and honours context cancellation: a cancelled or expired ctx aborts the drain promptly rather than buffering the entire child first.

func (*Eager) Next 

func (op *Eager) Next(out *Row) (bool, error)

Next emits the next buffered row.

func (*Eager) WithByteBudget added in v0.7.0 

func (op *Eager) WithByteBudget(maxBytes int64, estimateRow func(Row) int64) *Eager

WithByteBudget bounds the estimated retained size of the buffered rows by maxBytes. It complements the maxRows count cap so a few large-valued rows cannot exceed the engine's result-byte budget before the count cap fires (#1841). A non-positive maxBytes or nil estimateRow leaves the byte dimension disabled. Returns op for chaining and must be called before Init.

type EagerAggregation 

type EagerAggregation struct {
	// contains filtered or unexported fields
}

EagerAggregation is a blocking (pipeline-breaking) Volcano operator that groups rows from its child by the specified key columns and applies per-group aggregators. It emits one output row per group once the child is exhausted.

EagerAggregation is NOT safe for concurrent use.

func NewEagerAggregation 

func NewEagerAggregation(
	child Operator,
	keyCols []int,
	aggFactories []funcs.AggregatorFactory,
	maxGroups int,
) (*EagerAggregation, error)

NewEagerAggregation creates an EagerAggregation operator.

  • child: the upstream operator to consume.
  • keyCols: column indices whose values define the group key. An empty slice computes a single global aggregate.
  • aggFactories: one AggregatorFactory per aggregate expression. Must not be empty.
  • maxGroups: upper bound on distinct groups; pass 0 to use DefaultMaxGroups.

func (*EagerAggregation) Close 

func (op *EagerAggregation) Close() error

Close closes the child operator and releases internal state.

func (*EagerAggregation) Init 

func (op *EagerAggregation) Init(ctx context.Context) error

Init initialises the operator. The blocking consume phase is deferred to the first Next call.

func (*EagerAggregation) Next 

func (op *EagerAggregation) Next(out *Row) (bool, error)

Next emits the next aggregated row. On the first call it consumes all rows from the child (pipeline breaker) and builds the group table. Subsequent calls iterate through the completed groups.

func (*EagerAggregation) WithByteBudget added in v0.7.0 

func (op *EagerAggregation) WithByteBudget(maxBytes int64, estimateRow func(Row) int64) *EagerAggregation

WithByteBudget bounds the estimated retained size of the group keys by maxBytes. It complements the maxGroups count cap so a few large-valued group keys cannot exceed the engine's result-byte budget before the count cap fires (#1841). A non-positive maxBytes or nil estimateRow leaves the byte dimension disabled. Returns op for chaining and must be called before Init.

type Expand 

type Expand struct {
	// contains filtered or unexported fields
}

Expand is a Volcano pipeline operator that, for each input row, expands one hop along the graph's CSR adjacency.

Expand is NOT safe for concurrent use.

func NewExpand 

func NewExpand(input Operator, fwd, rev csrAdjacency, cfg ExpandConfig) *Expand

NewExpand creates an Expand operator. fwd is the forward CSR; rev is the reverse CSR (required for DirIn/DirBoth, ignored for DirOut).

func NewExpandWithOptions 

func NewExpandWithOptions(input Operator, fwd, rev csrAdjacency, cfg ExpandConfig, options ...expandOption) *Expand

NewExpandWithOptions creates an Expand operator with optional extra configuration applied via functional options (e.g. WithCyphermorphism).

All ExpandConfig fields behave identically to NewExpand. Options are applied after base construction and augment the operator's behaviour without altering its public type.

func (*Expand) Close 

func (op *Expand) Close() error

Close releases resources and closes the child operator.

func (*Expand) Init 

func (op *Expand) Init(ctx context.Context) error

Init initialises the operator and its child.

func (*Expand) Next 

func (op *Expand) Next(out *Row) (bool, error)

Next emits the next (srcID, edgeID, dstID) triplet appended to the current input row. It pulls a new input row whenever the current source's adjacency is exhausted.

type ExpandConfig 

type ExpandConfig struct {
	// Direction to follow. Defaults to DirOut when zero.
	Direction Direction
	// EdgeType, when non-empty, restricts emitted edges to those whose
	// positional index is present in EdgeTypeFilter with this type label.
	EdgeType string
	// EdgeTypeFilter maps absolute edge positions to type labels.  Required
	// when EdgeType is non-empty.
	EdgeTypeFilter map[uint64]string
	// InputCol is the column index in each input row that holds the source
	// NodeID (as expr.IntegerValue).  Defaults to 0.
	InputCol int
	// RelCols lists the input-row columns holding edge IDs already traversed
	// by sibling Expand operators in the same MATCH pattern. Each emitted
	// edge must NOT match any of these columns (openCypher 9 §3.2.2
	// relationship-isomorphism / cyphermorphism). Empty disables the
	// check.
	RelCols []int
	// MultiplicityFn returns the Cypher CREATE-call multiplicity recorded
	// for the directed edge (srcID, dstID). When the returned count is N >
	// 1, the operator emits the corresponding output row N times in a row,
	// reflecting the openCypher rule that `MATCH ()-[r]->()` enumerates
	// each CREATE call separately even when the underlying simple-graph
	// storage collapsed them to one entry (Merge5 [21]). A nil fn (or
	// returning 0 / 1) disables the multiplicity emit and behaves like a
	// plain single-row Expand.
	MultiplicityFn func(srcID, dstID uint64) int64
}

ExpandConfig carries the optional configuration for NewExpand.

type Filter 

type Filter struct {
	// contains filtered or unexported fields
}

Filter is a Volcano pipeline operator that applies a FilterFn predicate to each row produced by its child operator. It emits only rows for which the predicate returns expr.BoolValue(true).

Null and false both suppress the row (three-valued logic).

Filter is NOT safe for concurrent use.

func NewFilter 

func NewFilter(child Operator, predFn FilterFn) *Filter

NewFilter creates a Filter operator that wraps child and applies predFn to every row.

func (*Filter) Close 

func (op *Filter) Close() error

Close releases resources and closes the child operator.

func (*Filter) Init 

func (op *Filter) Init(ctx context.Context) error

Init initialises the operator and its child.

func (*Filter) Next 

func (op *Filter) Next(out *Row) (bool, error)

Next advances to the next row that passes the predicate. It pulls rows from the child until one satisfies the predicate, end-of-stream, or an error.

type FilterFn 

type FilterFn func(row Row) (expr.Value, error)

FilterFn is a predicate over a Row. It must return (BoolValue(true), nil) to accept the row. Any other non-error return value (including NULL and BoolValue(false)) causes the row to be dropped. An error halts the pipeline.

type Float64RangeIndex added in v0.6.0 

type Float64RangeIndex struct {
	// contains filtered or unexported fields
}

Float64RangeIndex adapts btree.Index[float64] to the [rangeLookup] interface for the UNIFIED numeric range seek (#1652). It backs a CREATE INDEX (btree) companion that indexes BOTH integer- and float-valued nodes under one float64 total order (openCypher orders integers and floats in a single numeric order), so a numeric range bound is a SUPERSET-complete probe over every numeric node — never a non-superset the way an int64-only index would be (it would silently drop the float-valued matches).

Both an integer and a float bound are coerced to float64: a Cypher IntegerValue bound and a FloatValue bound address the same numeric key space. The float64 widening of a large int64 bound can lose precision and make the probe over-return at the boundary, but the engine ALWAYS retains the original AST predicate as a residual Filter on top of the range scan, so any false positive is removed and the result is identical to a label scan+filter (cypher-expert-consultant, #1652). Nil / non-numeric bounds are treated as ±∞ so an unbounded side returns the whole numeric population.

NaN is never a bound here (the extractor only admits finite numeric literals/parameters) and is never indexed (projectNumericPropValue excludes it), so a NaN node can neither be a key nor be returned: the btree's total order places NaN below every real value and Range with a non-NaN lower bound never returns it, and the residual Filter is the final backstop.

func NewFloat64RangeIndex added in v0.6.0 

func NewFloat64RangeIndex(idx interface {
	Range(lo, hi float64) *roaring64.Bitmap
}) *Float64RangeIndex

NewFloat64RangeIndex constructs a Float64RangeIndex.

func (*Float64RangeIndex) RangeBitmap added in v0.6.0 

func (r *Float64RangeIndex) RangeBitmap(lo, hi expr.Value) *roaring64.Bitmap

RangeBitmap implements [rangeLookup]. A nil or NULL bound, or one that is neither an integer nor a float, is treated as the corresponding numeric infinity so an unbounded (or undecodable) side spans the whole numeric range — the residual Filter then enforces the exact predicate.

type GlobalAggregateAdapter 

type GlobalAggregateAdapter struct {
	// contains filtered or unexported fields
}

GlobalAggregateAdapter wraps an EagerAggregation operator that has no group-by keys and ensures the output stream contains exactly one row even when the child is empty.

GlobalAggregateAdapter is NOT safe for concurrent use.

func NewGlobalAggregateAdapter 

func NewGlobalAggregateAdapter(child Operator, aggFactories []funcs.AggregatorFactory) *GlobalAggregateAdapter

NewGlobalAggregateAdapter returns a GlobalAggregateAdapter that wraps child. aggFactories must contain one factory per aggregate column, in the same order as the columns produced by child. The factories are invoked exactly once, on empty input, to synthesise the neutral row.

Note: the adapter does not validate that child is a group-by-less EagerAggregation; callers are responsible for that. Wrapping a grouped aggregation has no effect because grouped aggregations always emit either zero (no input) or N rows (one per group), and the empty-input case for a grouped aggregation is correct as is.

func (*GlobalAggregateAdapter) Close 

func (op *GlobalAggregateAdapter) Close() error

Close releases the adapter's child.

func (*GlobalAggregateAdapter) Init 

func (op *GlobalAggregateAdapter) Init(ctx context.Context) error

Init initialises the adapter and its child.

func (*GlobalAggregateAdapter) Next 

func (op *GlobalAggregateAdapter) Next(out *Row) (bool, error)

Next forwards child rows unchanged. When the child reports end-of-stream without ever producing a row, Next emits exactly one synthetic row built from the neutral results of each aggregate factory before signalling exhaustion itself.

type GraphMutator 

type GraphMutator interface {
	// AddNode interns n and returns its stable NodeID. Returns the
	// error from the underlying graph implementation (currently only
	// [adjlist.ErrShardFull] is reachable, and only when the
	// underlying [adjlist.Config.MaxShardCapacity] is set).
	AddNode(n string) (graph.NodeID, error)

	// AddEdge inserts a directed edge (src→dst) with weight w, interning
	// endpoints as needed. Returns the stable NodeIDs of src and dst and
	// any error from the underlying graph implementation.
	AddEdge(src, dst string, w float64) (srcID, dstID graph.NodeID, err error)

	// AddEdgeH is [GraphMutator.AddEdge] with a stable per-edge handle
	// allocated by the underlying graph and stamped onto the adjacency
	// slot. The returned handle keys the *ByHandle metadata setters below
	// so a parallel CREATE's type and properties are resolvable on the
	// read path by an identity that survives sibling-edge deletion. Used
	// by CreateRelationship and by MergeRelationship's create path (rmp
	// #1683) so a MERGE-built parallel edge of a distinct type keeps its
	// own per-instance identity instead of collapsing onto the per-pair
	// label union. The handle is always non-zero on success.
	AddEdgeH(src, dst string, w float64) (srcID, dstID graph.NodeID, handle uint64, err error)

	// RemoveEdge removes the directed edge from src to dst (no-op if absent).
	RemoveEdge(src, dst string)

	// SetNodeLabel attaches label to n (inserting n if absent). Returns
	// any error from the underlying [adjlist.AdjList.AddNode] (see
	// [GraphMutator.AddNode]).
	SetNodeLabel(n, label string) error

	// RemoveNodeLabel detaches label from n (no-op if absent).
	RemoveNodeLabel(n, label string)

	// RemoveNode tombstones n in the underlying graph so subsequent reads
	// (AllNodesScan, count(*), Order) treat the node as absent. Callers
	// should strip labels/properties/incident edges before invoking
	// RemoveNode so the tombstone reflects the fully-deleted state.
	RemoveNode(n string)

	// IsTombstoned reports whether the NodeID has been tombstoned. Used by
	// AllNodesScan to skip phantom entries the Mapper still indexes.
	IsTombstoned(id graph.NodeID) bool

	// SetNodeProperty sets the named property on n. Returns any error
	// from the underlying [adjlist.AdjList.AddNode] (see
	// [GraphMutator.AddNode]).
	SetNodeProperty(n, key string, value lpg.PropertyValue) error

	// DelNodeProperty removes the named property from n (no-op if absent).
	DelNodeProperty(n, key string)

	// NodeProperties returns a snapshot of all properties currently on n.
	NodeProperties(n string) map[string]lpg.PropertyValue

	// NodeLabels returns a snapshot of all labels currently on n in
	// unspecified order.
	NodeLabels(n string) []string

	// HasEdge reports whether a directed edge from src to dst is present.
	HasEdge(src, dst string) bool

	// SetEdgeLabel attaches label to the directed edge (src, dst).
	SetEdgeLabel(src, dst, label string)

	// SetEdgeProperty sets the named property on the directed edge (src, dst).
	// Returns any error from the underlying graph (e.g. schema violation).
	SetEdgeProperty(src, dst, key string, value lpg.PropertyValue) error

	// DelEdgeProperty removes the named property from the directed edge
	// (src, dst) (no-op if absent).
	DelEdgeProperty(src, dst, key string)

	// EdgeProperties returns a snapshot of every property currently set on
	// the directed edge (src, dst). Returns an empty map when the edge has
	// no properties or does not exist.
	EdgeProperties(src, dst string) map[string]lpg.PropertyValue

	// EdgeLabels returns a snapshot of every label currently attached to
	// the directed edge (src, dst). Returns an empty slice when the edge
	// has no labels or does not exist. Used by DELETE r to capture the
	// relationship type before tombstoning the edge so the row's
	// post-delete RelationshipValue keeps `RETURN type(r)` working.
	EdgeLabels(src, dst string) []string

	// IncEdgeCreateCount bumps the Cypher CREATE-call multiplicity
	// counter for the directed edge (src, dst) by one and returns the
	// new (1-based) value. The counter records how many CREATE
	// statements have targeted the same endpoint pair regardless of
	// whether the underlying storage already had an entry — MERGE
	// consults it to emit multiplicity rows when an existing edge
	// satisfies the merge pattern (Merge5 [3]). The returned index is
	// the per-instance idx callers pass to the *At family of
	// metadata-write helpers.
	IncEdgeCreateCount(src, dst string) int64
	// EdgeCreateCount returns the current CREATE-call multiplicity
	// counter for the directed edge (src, dst), or 0 when no CREATE
	// has been recorded.
	EdgeCreateCount(src, dst string) int64
	// DecEdgeCreateCount decrements the CREATE-call multiplicity
	// counter (floor 0). Called by DELETE so subsequent MERGEs see
	// the correct multiplicity.
	DecEdgeCreateCount(src, dst string)
	// SetEdgeLabelAt attaches `label` to the directed edge instance
	// (src, dst) at the supplied 1-based CREATE index. Used by
	// CreateRelationship so parallel CREATEs of the same endpoint
	// pair retain their distinct labels (Match2 [6] / Match7 [29]).
	SetEdgeLabelAt(src, dst string, idx int64, label string)
	// EdgeLabelsAt returns the labels recorded at instance `idx` of
	// the directed edge (src, dst), or nil when the instance has no
	// per-CREATE labels.
	EdgeLabelsAt(src, dst string, idx int64) []string
	// SetEdgePropertyAt records `key`=`value` on the directed edge
	// instance (src, dst) at the supplied 1-based CREATE index. Returns
	// any error returned by the installed SchemaValidator.
	SetEdgePropertyAt(src, dst string, idx int64, key string, value lpg.PropertyValue) error
	// EdgePropertiesAt returns the property map recorded at instance
	// `idx` of the directed edge (src, dst), or nil when no
	// per-CREATE map was captured.
	EdgePropertiesAt(src, dst string, idx int64) map[string]lpg.PropertyValue
	// RemoveEdgeInstance drops every per-CREATE label and property
	// associated with (src, dst) at `idx`. Used by DELETE to discard
	// a specific logical edge while leaving sibling instances
	// untouched.
	RemoveEdgeInstance(src, dst string, idx int64)

	// SetEdgeLabelByHandle attaches `label` to the edge identified by the
	// stable `handle` on the (src, dst) pair (see [GraphMutator.AddEdgeH]).
	// The handle-keyed analogue of SetEdgeLabelAt; the read path resolves a
	// parallel CREATE's type by this identity instead of a positional CSR
	// index. No-op when handle is 0.
	SetEdgeLabelByHandle(src, dst string, handle uint64, label string)
	// EdgeLabelsByHandle returns the labels recorded for the edge
	// identified by `handle` on the (src, dst) pair, or nil when none.
	EdgeLabelsByHandle(src, dst string, handle uint64) []string
	// SetEdgePropertyByHandle records `key`=`value` on the edge identified
	// by the stable `handle` on the (src, dst) pair. No-op when handle is 0.
	// Returns any error returned by the installed SchemaValidator.
	SetEdgePropertyByHandle(src, dst string, handle uint64, key string, value lpg.PropertyValue) error
	// DelEdgePropertyByHandle removes exactly `key` from the property bag of
	// the edge identified by the stable `handle` on the (src, dst) pair,
	// leaving sibling handles untouched. No-op when handle is 0 or the handle
	// never carried the key. The single-key removal analogue of
	// RemoveEdgeInstanceByHandle (which drops ALL of a handle's metadata); used
	// by REMOVE r.x / SET r.x = null on one parallel relationship instance.
	DelEdgePropertyByHandle(src, dst string, handle uint64, key string)
	// EdgePropertiesByHandle returns the property map recorded for the edge
	// identified by `handle` on the (src, dst) pair, or nil when none.
	EdgePropertiesByHandle(src, dst string, handle uint64) map[string]lpg.PropertyValue
	// RemoveEdgeInstanceByHandle drops every per-handle label and property
	// associated with (src, dst) at `handle`. Used by DELETE to discard a
	// specific logical edge while leaving sibling handles untouched.
	RemoveEdgeInstanceByHandle(src, dst string, handle uint64)

	// EdgeHandleAtPosition resolves the stable per-edge handle of the bound
	// relationship instance whose forward-CSR edge position is `edgePos` (the
	// edge-position counter the Expand operator places in a relationship
	// variable's column). It returns the exact handle of that one parallel
	// instance, or 0 when no handle is resolvable — the edge carries no stable
	// handle (simple-graph / pre-handle storage), either endpoint is unknown,
	// the position is out of range, or the slot does not point at dst. Callers
	// that get 0 must fall back to the per-pair store only and must NOT mutate
	// any by-handle instance, so a SET/REMOVE never corrupts the wrong instance.
	//
	// It is the write-path counterpart of the read path's position→handle
	// resolution and reads the same per-query forward-CSR snapshot, so it is
	// exact for parallel edges and consistent with how reads identify the bound
	// instance.
	EdgeHandleAtPosition(src, dst string, edgePos uint64) uint64

	// FirstEdgeHandle returns the stable per-edge handle stamped on the FIRST
	// adjacency slot from src to dst, and whether such a handled slot exists.
	// It is the by-PAIR handle resolver (not positional like
	// [GraphMutator.EdgeHandleAtPosition]): MERGE binds a single logical
	// (src, dst) edge rather than a specific parallel instance, so the
	// ON MATCH / ON CREATE action path uses it to mirror its per-pair property
	// writes onto the matched edge's by-handle store. Returns 0 / false when
	// either endpoint is unknown, no src→dst edge exists, or the matched slot
	// carries the 0 "no handle" sentinel (simple-graph / pre-handle storage) —
	// in which case the caller mutates the per-pair store only and never a
	// by-handle instance.
	FirstEdgeHandle(src, dst string) (uint64, bool)

	// OutNeighbours returns the outgoing neighbour node keys of n as a
	// snapshot slice. Callers must not mutate the returned slice.
	OutNeighbours(n string) []string

	// InNeighbours returns the incoming neighbour node keys of n as a
	// snapshot slice. This requires a full graph walk for directed adjacency
	// lists that do not maintain a reverse index. Callers must not mutate the
	// returned slice.
	InNeighbours(n string) []string

	// RemoveAllEdgesFrom removes all edges incident from n in O(degree) time
	// and returns the outgoing neighbour keys that were removed. For undirected
	// graphs the mirror entries are also removed. The per-pair edge state
	// (labels, properties, handle metadata, CREATE counters) is cleared for
	// every removed pair, exactly as a sequence of RemoveEdge calls would do.
	//
	// Callers that also need to roll back undo-log entries or emit WAL records
	// must capture the neighbours via OutNeighbours before calling this method;
	// RemoveAllEdgesFrom does not emit WAL records on its own.
	RemoveAllEdgesFrom(n string)

	// OutDegree returns the number of outgoing edges from n.
	OutDegree(n string) int

	// ResolveNodeID translates a user-facing node key to its internal NodeID,
	// returning ok=false when the node has not been interned yet.
	ResolveNodeID(n string) (graph.NodeID, bool)

	// ResolveNodeLabel translates an internal NodeID back to the user-facing
	// node key, returning ok=false when id is unknown.
	ResolveNodeLabel(id graph.NodeID) (string, bool)

	// WalkNodeIDs calls fn for every node currently interned in the graph.
	WalkNodeIDs(fn func(graph.NodeID) bool)
}

GraphMutator is the write surface exposed to Cypher write operators.

All methods accept the user-facing node key (string) used by the lpg.Graph[string,float64] instantiation. The graph is responsible for interning the value and returning the stable internal NodeID where applicable.

GraphMutator is NOT safe for concurrent use from multiple goroutines; each physical operator tree owns exactly one instance.

type HashJoin added in v0.3.1 

type HashJoin struct {
	// contains filtered or unexported fields
}

HashJoin is a Volcano pipeline operator that performs an order-insensitive equi-join between two independent (uncorrelated) plans. It replaces a nested-loop Apply + equi-join Filter when the planner has proven the substitution is result-identical and order-safe.

HashJoin is NOT safe for concurrent use.

func NewHashJoin added in v0.3.1 

func NewHashJoin(build, probe Operator, buildFn, probeFn KeyFn, buildOnLeft bool) *HashJoin

NewHashJoin creates a HashJoin.

  • build is the plan whose rows are fully materialised into the hash table.
  • probe is the plan streamed against the table.
  • buildFn / probeFn extract the join key from a build / probe row.
  • buildOnLeft selects the output column order (see HashJoin.buildOnLeft).

HashJoin takes ownership of both plans; callers must not use them afterwards.

func (*HashJoin) Close added in v0.3.1 

func (op *HashJoin) Close() error

Close releases the hash table and closes both child plans.

func (*HashJoin) Init added in v0.3.1 

func (op *HashJoin) Init(ctx context.Context) error

Init initialises both child plans and resets join state.

func (*HashJoin) Next added in v0.3.1 

func (op *HashJoin) Next(out *Row) (bool, error)

Next advances the hash join.

func (*HashJoin) WithByteBudget added in v0.7.0 

func (op *HashJoin) WithByteBudget(maxBytes int64, estimateRow func(Row) int64) *HashJoin

WithByteBudget bounds the estimated retained size of the build table by maxBytes, returning ErrHashJoinMemoryExceeded when exceeded. The build table has no count cap, so this is the operator's memory bound (#1841). A non-positive maxBytes or nil estimateRow leaves it unbounded (prior behaviour). Returns op for chaining and must be called before Init.

type IndexBuffer 

type IndexBuffer struct {
	// contains filtered or unexported fields
}

IndexBuffer collects index.Change events produced by write operators during a single write transaction.

Call Enqueue for every graph mutation. At the transaction boundary:

  • Commit: fans changes to index.Manager.ApplyBatch then resets.
  • Rollback: discards changes without touching indexes.

IndexBuffer is NOT safe for concurrent use.

func (*IndexBuffer) Commit 

func (b *IndexBuffer) Commit(mgr *index.Manager)

Commit applies all buffered changes to mgr via ApplyBatch, then resets the buffer. A nil mgr is safe: changes are discarded without panicking.

func (*IndexBuffer) Enqueue 

func (b *IndexBuffer) Enqueue(c index.Change)

Enqueue appends c to the buffer.

func (*IndexBuffer) Len 

func (b *IndexBuffer) Len() int

Len returns the number of changes currently buffered.

func (*IndexBuffer) Rollback 

func (b *IndexBuffer) Rollback()

Rollback discards all buffered changes without applying them.

type IndexKindExec 

type IndexKindExec uint8

IndexKindExec distinguishes hash vs. btree in the exec layer. 

const (
	// ExecIndexHash creates a hash.Index[string].
	ExecIndexHash IndexKindExec = iota
	// ExecIndexBTree creates a btree.Index[string].
	ExecIndexBTree
)

type Int64HashIndex 

type Int64HashIndex struct {
	// contains filtered or unexported fields
}

Int64HashIndex adapts hash.Index[int64] to the [hashLookup] interface. It accepts only expr.IntegerValue seek keys.

func NewInt64HashIndex 

func NewInt64HashIndex(idx interface {
	LookupAppend(value int64, dst []uint64) []uint64
}) *Int64HashIndex

NewInt64HashIndex constructs an Int64HashIndex.

func (*Int64HashIndex) LookupAppend added in v0.6.0 

func (h *Int64HashIndex) LookupAppend(value expr.Value, dst []uint64) ([]uint64, error)

LookupAppend implements [hashLookup].

type Int64RangeIndex 

type Int64RangeIndex struct {
	// contains filtered or unexported fields
}

Int64RangeIndex adapts btree.Index[int64] to the [rangeLookup] interface. Nil bounds are treated as ±∞ using math.MinInt64 / math.MaxInt64.

func NewInt64RangeIndex 

func NewInt64RangeIndex(idx interface {
	Range(lo, hi int64) *roaring64.Bitmap
}) *Int64RangeIndex

NewInt64RangeIndex constructs an Int64RangeIndex.

func (*Int64RangeIndex) RangeBitmap 

func (r *Int64RangeIndex) RangeBitmap(lo, hi expr.Value) *roaring64.Bitmap

RangeBitmap implements [rangeLookup].

type KeyFn added in v0.3.1 

type KeyFn func(row Row) (expr.Value, error)

KeyFn extracts the join-key value from a row. It returns the evaluated key (which may be expr.Null) or an error that halts the pipeline.

type LPGLabelSource 

type LPGLabelSource struct {
	// contains filtered or unexported fields
}

LPGLabelSource is a concrete [labelResolver] built from an lpg label registry and a label index. It is the standard adapter for tests and production use.

func NewLPGLabelSource 

func NewLPGLabelSource(reg *label.Index, lookupFn func(string) (uint32, bool)) *LPGLabelSource

NewLPGLabelSource constructs a LPGLabelSource. lookupFn should wrap lpg.LabelRegistry.Lookup, casting LabelID to uint32.

func (*LPGLabelSource) ResolveLabelBitmap 

func (s *LPGLabelSource) ResolveLabelBitmap(name string) *roaring64.Bitmap

ResolveLabelBitmap implements [labelResolver].

type Limit 

type Limit struct {
	// contains filtered or unexported fields
}

Limit is a Volcano pipeline operator that forwards at most n rows from its child operator and then signals end-of-stream.

Limit is NOT safe for concurrent use.

func NewLimit 

func NewLimit(child Operator, n int64) (*Limit, error)

NewLimit creates a Limit operator that passes at most n rows from child. n must be ≥ 0; a limit of 0 emits no rows.

func (*Limit) Close 

func (op *Limit) Close() error

Close releases resources and closes the child operator.

func (*Limit) Init 

func (op *Limit) Init(ctx context.Context) error

Init initialises the operator and resets the emission counter.

func (*Limit) Next 

func (op *Limit) Next(out *Row) (bool, error)

Next forwards the next row from the child, unless the limit has been reached, in which case it returns (false, nil) immediately.

type Merge 

type Merge struct {
	// contains filtered or unexported fields
}

Merge implements MERGE semantics: match-or-create a pattern.

Merge is NOT safe for concurrent use.

func NewMerge 

func NewMerge(
	nodeVar string,
	labels []string,
	properties string,
	onCreateStrs, onMatchStrs []string,
	searchFn MergeSearchFn,
	schema map[string]int,
	child Operator,
	mutator GraphMutator,
) (*Merge, error)

NewMerge creates a Merge operator.

nodeVar is the variable bound to the merged node. labels and properties are the node-pattern components used when creating a new node. onCreateStrs and onMatchStrs are opaque SET-item strings from the IR translator. searchFn executes the read side of the match. schema maps variable names to column indices. mutator is the graph write surface.

func (*Merge) Close 

func (op *Merge) Close() error

Close closes the child operator.

func (*Merge) Init 

func (op *Merge) Init(ctx context.Context) error

Init initialises the operator: executes the search plan, then dispatches to the ON MATCH or ON CREATE branch depending on whether the search returned any rows.

The first Merge.Init (or CreateNode.Init) in the process also seeds [globalNodeCounter] past the largest synthetic key already interned in op.mutator, so that the keys minted by [Merge.freshNodeKey] in this process cannot collide with __cx_merge_<hex> keys persisted by an earlier process and replayed during WAL / snapshot recovery. Without this a one-process-per-command consumer mints __cx_merge_1 on every command and the second MERGE silently overwrites the first node. The seed is gated by [globalNodeCounterSeededOnce] so the O(N) scan runs at most once per process regardless of how many CreateNode / Merge operators are built.

func (*Merge) Next 

func (op *Merge) Next(out *Row) (bool, error)

Next emits one row: either a matched row (ON MATCH) or the created row (ON CREATE), each emitted exactly once.

func (*Merge) WithActionEvals added in v0.8.0 

func (op *Merge) WithActionEvals(onCreate, onMatch map[string]ValueEvalFn) *Merge

WithActionEvals attaches per-row RHS evaluators for ON CREATE / ON MATCH property-set items whose right-hand side is a non-literal expression (keyed by MergeActionEvalKey). Without these, a self-referential assignment such as `ON MATCH SET n.num = n.num + 1` fails to parse as a literal and is silently dropped (#1965). Returns op for chaining.

func (*Merge) WithConstraints 

func (op *Merge) WithConstraints(reg *ConstraintRegistry, mgr *index.Manager) *Merge

WithConstraints attaches a ConstraintRegistry and index.Manager for pre-write enforcement in ON CREATE and ON MATCH actions. Both must be non-nil. Returns op for chaining.

func (*Merge) WithParams 

func (op *Merge) WithParams(params map[string]expr.Value) (*Merge, error)

WithParams re-parses the property map with the supplied query parameters for $name substitution. Returns op for chaining.

func (*Merge) WithPropsEvalFn 

func (op *Merge) WithPropsEvalFn(fn PropsEvalFn) *Merge

WithPropsEvalFn attaches a per-row property evaluator. When fn is non-nil the operator re-evaluates the MERGE node-pattern property map against each driving row and uses the merged (literal ∪ dynamic) property set both as the search predicate and as the ON CREATE node-property writes. Required for MERGE patterns whose inline property map contains variable references such as `MERGE (p:Person {login: prop.login})` after an UNWIND.

Returns op for chaining.

type MergePattern added in v0.7.0 

type MergePattern struct {
	// contains filtered or unexported fields
}

MergePattern matches-or-creates a chain of one or more relationship hops in which at least one node is not already bound. See the package doc above for the full algorithm.

func NewMergePattern added in v0.7.0 

func NewMergePattern(child Operator, mutator GraphMutator) *MergePattern

NewMergePattern creates an empty MergePattern; call AddBoundNode/ AddFreshNode once per chain position (in order) and AddHop once per relationship (len(hops) == positions added - 1), then WithActions.

func (*MergePattern) AddBoundNode added in v0.7.0 

func (op *MergePattern) AddBoundNode(varName string, col int) *MergePattern

AddBoundNode appends a chain position whose value is already bound by an earlier clause, read from input-row column col.

func (*MergePattern) AddFreshNode added in v0.7.0 

func (op *MergePattern) AddFreshNode(varName string, labels []string, propsRaw string, outCol int) *MergePattern

AddFreshNode appends a chain position this MERGE clause introduces fresh. outCol is the output-row column that receives the matched-or-created NodeID as an expr.IntegerValue; pass -1 if the variable is never referenced downstream (still searched/created, just not projected).

func (*MergePattern) AddHop added in v0.7.0 

func (op *MergePattern) AddHop(relVar string, relCol int, relType, relPropsRaw string, undirected, reversed bool) *MergePattern

AddHop appends the relationship connecting the two most-recently-added chain positions. relCol is the output-row column for a named relationship variable, or -1 for an anonymous one. reversed is true for an incoming `<-` pattern, where the edge is stored from the next chain position back to the current one.

func (*MergePattern) Close added in v0.7.0 

func (op *MergePattern) Close() error

Close closes the child operator.

func (*MergePattern) Init added in v0.7.0 

func (op *MergePattern) Init(ctx context.Context) error

Init initialises the operator and its child. The first MergePattern.Init (or CreateNode.Init / Merge.Init) in the process also seeds [globalNodeCounter], exactly as those operators do, so fresh-node keys minted here cannot collide with __cx_merge_<hex> keys replayed from an earlier process during WAL / snapshot recovery.

func (*MergePattern) Next added in v0.7.0 

func (op *MergePattern) Next(out *Row) (bool, error)

Next drives one search-or-create cycle per input row, buffering multiple matches (Merge5-style multiplicity, generalised: a joint pattern can have more than one satisfying binding) so each is emitted as its own row. Mirrors Merge.Next's buffering shape, including firing exactly once against an empty driving row when MergePattern is the leading clause (no preceding MATCH/WITH at all).

func (*MergePattern) WithActionEvals added in v0.8.0 

func (op *MergePattern) WithActionEvals(onCreate, onMatch map[string]ValueEvalFn) *MergePattern

WithActionEvals attaches per-row RHS evaluators for ON CREATE / ON MATCH property-set items whose right-hand side is a non-literal expression (keyed by MergeActionEvalKey on the target variable and property key). Without these, an expression such as `ON MATCH SET n.num = n.num + 1` fails to parse as a literal and is dropped (#1965). Returns op for chaining.

func (*MergePattern) WithActions added in v0.7.0 

func (op *MergePattern) WithActions(onCreateStrs, onMatchStrs []string) (*MergePattern, error)

WithActions parses and attaches ON CREATE / ON MATCH SET items (the same opaque-string representation Merge uses). Each item may target any variable in the chain (a fresh node, a bound node, or a hop's relationship variable) — dispatched at apply time by the parsed action's target name.

func (*MergePattern) WithConstraints added in v0.7.0 

func (op *MergePattern) WithConstraints(reg *ConstraintRegistry, mgr *index.Manager) *MergePattern

WithConstraints attaches a ConstraintRegistry and index.Manager for pre-write enforcement of every fresh node's created properties and of every ON CREATE / ON MATCH node property action — mirroring Merge.WithConstraints and CreateNode.WithConstraints. Both must be non-nil. Returns op for chaining.

func (*MergePattern) WithNodePropsEvalFn added in v0.7.0 

func (op *MergePattern) WithNodePropsEvalFn(fn PropsEvalFn) *MergePattern

WithNodePropsEvalFn attaches a per-row property evaluator to the most-recently-added fresh node (mirroring [CreateNode.WithPropsEvalFn| CreateNode.WithPropsEvalFn]). Used when that node's property map contains non-literal expressions (parameters, variable references) that [parsePropLiteral] cannot resolve at plan-construction time; the dynamic entries are merged with the map's literal entries at both search and create time (see [mergeProps]), taking precedence on key collision. Without this, a node position's parameterised properties would be silently dropped from both the search predicate and the created node — the same defect class this operator exists to eliminate.

func (*MergePattern) WithParams added in v0.7.0 

func (op *MergePattern) WithParams(params map[string]expr.Value) (*MergePattern, error)

WithParams attaches query parameters for $name substitution in every node's and every hop's inline property map, re-parsing each non-empty raw map with parameter references resolved to concrete literal values. Mirrors CreateNode.WithParams: resolving parameters once here, at build time, is cheaper than a per-row PropsEvalFn and is correct because parameter values are constant for the whole query execution. Must be called after every AddBoundNode/AddFreshNode/AddHop call it should cover.

type MergeRelAction 

type MergeRelAction struct {
	// contains filtered or unexported fields
}

MergeRelAction is a pre-parsed `SET <relVar>.<key> = <value>` item, or a whole-entity REPLACE sentinel (#1687).

Three shapes:

  • Single-property write: key != "", value is the literal string.
  • Entity-copy: key == "", value == "<sourceVar>". When replace is true the edge's properties absent from the source entity are cleared first.
  • Replace-map sentinel: key == "", value == "", replace == true. retainKeys lists the RHS map keys; the edge's properties absent from retainKeys are cleared. The sentinel is immediately followed by the per-key write actions for the map, so the clear precedes the writes. An empty (non-nil) retainKeys clears every property (`SET r = {}`).

func MergeRelActionFromKV 

func MergeRelActionFromKV(key, value string) MergeRelAction

MergeRelActionFromKV constructs a MergeRelationship ON CREATE / ON MATCH action from a (key, value) pair. value is the opaque literal string as it appears in the source query (e.g. `'foo'` or `42`).

func MergeRelActionReplaceFromKV added in v0.6.0 

func MergeRelActionReplaceFromKV(key, value string, replace bool, retainKeys []string) MergeRelAction

MergeRelActionReplaceFromKV constructs a MergeRelationship ON CREATE / ON MATCH action carrying the whole-entity REPLACE marker (#1687). When replace is true the action is either a replace-map sentinel (key == "", value == "", retainKeys lists the RHS keys to keep) or an entity-copy replace (key == "", value == "<sourceVar>", retainKeys nil → retain the source's live keys). retainKeys is copied defensively so the caller may reuse its slice. When replace is false this is equivalent to MergeRelActionFromKV.

type MergeRelationship 

type MergeRelationship struct {
	// contains filtered or unexported fields
}

MergeRelationship matches-or-creates a single-hop directed relationship between two already-bound endpoint columns. ON CREATE / ON MATCH actions targeting the relationship variable are applied to the matched-or-created edge.

MergeRelationship is NOT safe for concurrent use.

func NewMergeRelationship 

func NewMergeRelationship(child Operator, srcCol, dstCol int, relType string, mutator GraphMutator) *MergeRelationship

NewMergeRelationship constructs a MergeRelationship operator.

  • child is the upstream plan providing rows with the bound endpoints.
  • srcCol / dstCol are the column indices that hold the src / dst NodeID.
  • relType is the relationship type label (single label only).
  • mutator is the graph write surface.

func (*MergeRelationship) Close 

func (op *MergeRelationship) Close() error

Close closes the child operator.

func (*MergeRelationship) Init 

func (op *MergeRelationship) Init(ctx context.Context) error

Init initialises the operator and its child.

func (*MergeRelationship) Next 

func (op *MergeRelationship) Next(out *Row) (bool, error)

Next emits the next input row, ensuring that the (src)-[:relType]->(dst) edge exists in the graph (either pre-existing or newly created). When an existing edge has CREATE-multiplicity N > 1 the operator emits N rows for the same upstream tuple (Merge5 [3]).

func (*MergeRelationship) WithActionEvals added in v0.8.0 

func (op *MergeRelationship) WithActionEvals(onCreate, onMatch map[string]ValueEvalFn) *MergeRelationship

WithActionEvals attaches per-row RHS evaluators for ON CREATE / ON MATCH property-set items whose right-hand side is a non-literal expression (keyed by MergeActionEvalKey on the relationship variable and property key). Without these, `ON MATCH SET r.n = r.n + 1` fails to parse as a literal and, on this fast path, surfaced a parse error instead of incrementing the edge property (#1965). Returns op for chaining.

func (*MergeRelationship) WithOnCreate 

func (op *MergeRelationship) WithOnCreate(relVar string, actions []MergeRelAction) *MergeRelationship

WithOnCreate registers ON CREATE SET actions to apply when the edge is newly created. Each action is `<relVar>.<key> = <value>`; the caller has already verified that every action targets the relationship variable bound by this operator.

func (*MergeRelationship) WithOnMatch 

func (op *MergeRelationship) WithOnMatch(relVar string, actions []MergeRelAction) *MergeRelationship

WithOnMatch registers ON MATCH SET actions to apply when the edge already exists.

func (*MergeRelationship) WithRelColumn 

func (op *MergeRelationship) WithRelColumn(relCol int) *MergeRelationship

WithRelColumn registers the output-row column index that will carry the matched / created edge ID. When set (relCol >= 0) MergeRelationship extends the row with an IntegerValue(edgeID) at the column so downstream operators (RETURN r, count(r), …) see the bound relationship.

func (*MergeRelationship) WithRelProperties 

func (op *MergeRelationship) WithRelProperties(propsRaw string) *MergeRelationship

WithRelProperties registers an inline relationship property predicate (e.g. `{name: 'r2'}` from `MERGE (a)-[r:T {name: 'r2'}]->(b)`). When set, the operator filters the existing-edge search by the predicate AND writes the listed properties when a new edge is created. Pass an empty string to clear.

func (*MergeRelationship) WithSchema 

func (op *MergeRelationship) WithSchema(schema map[string]int) *MergeRelationship

WithSchema attaches the upstream variable-to-column mapping so entity-copy actions (`SET r = a`) can resolve the source variable from the row at write time.

func (*MergeRelationship) WithUndirected 

func (op *MergeRelationship) WithUndirected(u bool) *MergeRelationship

WithUndirected toggles the undirected-search behaviour. When true, the match phase probes both (src, dst) and (dst, src) directions before falling through to the edge-create path, matching the openCypher semantics of `MERGE (a)-[r:T]-(b)` (Merge5 [13]).

type MergeSearchFn 

type MergeSearchFn func(ctx context.Context) ([]Row, error)

MergeSearchFn is a function that executes the search sub-plan for the MERGE pattern and returns the matching rows. An empty slice means no match.

func NewMergeSearchFnFromPattern 

func NewMergeSearchFnFromPattern(
	labels []string,
	propertiesRaw string,
	params map[string]expr.Value,
	mutator GraphMutator,
) (MergeSearchFn, error)

NewMergeSearchFnFromPattern returns a MergeSearchFn that finds every node in mutator whose label set contains every label in labels and whose property bag is equal to every (key, value) parsed from propertiesRaw.

labels is the slice of pattern labels (may be empty when the pattern is e.g. `(n {key: v})`). propertiesRaw is the opaque literal-map string surfaced by the IR (e.g. `{name: "Alice", age: 30}`); it may be empty. params binds `$name` references in propertiesRaw to query parameters; when empty the parser ignores parameter substitution.

The function returned by NewMergeSearchFnFromPattern walks every interned node id, resolves the label and property bag, and admits the node iff every label and every property matches. Match scaling is O(N) where N is the number of interned nodes; the cost is acceptable for the typical MERGE workload (small N or label-restricted pattern). A future revision may use [labelResolver]'s bitmap intersection to short-circuit label scans.

type NodeByIndexRangeScan 

type NodeByIndexRangeScan struct {
	// contains filtered or unexported fields
}

NodeByIndexRangeScan is a Volcano leaf operator that scans a B+tree index over a half-open, closed, or open interval. Each Row has a single column: expr.IntegerValue(nodeID).

NodeByIndexRangeScan is NOT safe for concurrent use.

func NewNodeByIndexRangeScan 

func NewNodeByIndexRangeScan(idx rangeLookup, lo, hi RangeBound) *NodeByIndexRangeScan

NewNodeByIndexRangeScan creates a NodeByIndexRangeScan.

func (*NodeByIndexRangeScan) Close 

func (op *NodeByIndexRangeScan) Close() error

Close releases resources.

func (*NodeByIndexRangeScan) Init 

func (op *NodeByIndexRangeScan) Init(ctx context.Context) error

Init performs the range lookup and initialises the bitmap iterator.

func (*NodeByIndexRangeScan) Next 

func (op *NodeByIndexRangeScan) Next(out *Row) (bool, error)

Next emits the next NodeID in the inclusive [lo, hi] superset. Returns (false, nil) at end-of-stream. Exact open/closed enforcement is the caller's residual-filter responsibility (see the type doc); this operator emits every NodeID the index's range bitmap contains.

type NodeByIndexSeek 

type NodeByIndexSeek struct {
	// contains filtered or unexported fields
}

NodeByIndexSeek is a Volcano leaf operator that performs an equality lookup on a property hash index. Each Row has a single column: expr.IntegerValue(nodeID).

NodeByIndexSeek is NOT safe for concurrent use.

func NewNodeByIndexSeek 

func NewNodeByIndexSeek(idx hashLookup, seekValue expr.Value) *NodeByIndexSeek

NewNodeByIndexSeek creates a NodeByIndexSeek that looks up seekValue in idx.

func (*NodeByIndexSeek) Close 

func (op *NodeByIndexSeek) Close() error

Close releases resources.

func (*NodeByIndexSeek) Init 

func (op *NodeByIndexSeek) Init(ctx context.Context) error

Init performs the index lookup, draining the matching NodeIDs into the operator's reused buffer. The dominant singleton/small posting list fits the inline idbuf, so a seek allocates nothing after the buffer is established.

func (*NodeByIndexSeek) Next 

func (op *NodeByIndexSeek) Next(out *Row) (bool, error)

Next emits the next matching NodeID. Returns (false, nil) at end-of-stream.

type NodeByLabelScan 

type NodeByLabelScan struct {
	// contains filtered or unexported fields
}

NodeByLabelScan is a Volcano leaf operator that emits one Row per NodeID carrying the named label. Each Row has a single column: expr.IntegerValue(nodeID).

NodeByLabelScan is NOT safe for concurrent use.

func NewNodeByLabelScan 

func NewNodeByLabelScan(labelName string, src labelResolver) *NodeByLabelScan

NewNodeByLabelScan creates a NodeByLabelScan for the given label.

func (*NodeByLabelScan) Close 

func (op *NodeByLabelScan) Close() error

Close releases resources held by the operator.

func (*NodeByLabelScan) Init 

func (op *NodeByLabelScan) Init(ctx context.Context) error

Init resolves the label to a bitmap and initialises the iterator.

func (*NodeByLabelScan) Next 

func (op *NodeByLabelScan) Next(out *Row) (bool, error)

Next emits the next matching NodeID. Returns (false, nil) at end-of-stream.

type Operator 

type Operator interface {
	// Init initialises the operator and its children. ctx is stored for later
	// use in Next; implementations must not begin producing rows in Init.
	Init(ctx context.Context) error

	// Next advances the operator by one row, writing the result into out.
	// It returns (true, nil) if a row was written, (false, nil) at end-of-stream,
	// or (false, err) on error. After returning (false, _), Next must not be
	// called again.
	//
	// Implementations check ctx.Done() on every call. Long-running loops check
	// ctx.Done() every 4096 iterations.
	Next(out *Row) (bool, error)

	// Close releases all resources held by this operator (open file handles,
	// memory, goroutines). It must be called exactly once by the pipeline
	// driver, even when Next returned an error.
	Close() error
}

Operator is the core abstraction of the Volcano iterator model. Every node in a physical query plan implements this interface.

Lifecycle

  1. [Init] is called exactly once before the first call to [Next].
  2. [Next] is called repeatedly until it returns (false, nil) or an error.
  3. [Close] is called exactly once, regardless of whether [Next] returned an error. Implementations must release all resources in [Close].

Cancellation

Every [Next] implementation must check ctx.Done() at the top of the call. For long-running inner loops that produce more than 4096 tuples without returning, check ctx.Done() every 4096 iterations.

Concurrency

An Operator instance is NOT safe for concurrent use. Each goroutine in a parallel pipeline segment owns its own operator tree.

type OptionalApply 

type OptionalApply struct {
	// contains filtered or unexported fields
}

OptionalApply is the left-outer variant of CorrelatedApply. For every outer row, the inner pipeline is driven exactly like CorrelatedApply; when the inner pipeline produces zero rows for a given outer row, OptionalApply emits a single NULL-extended row whose width equals the configured padded width, holding the outer columns followed by NULL placeholders for the inner-introduced columns.

OptionalApply is NOT safe for concurrent use.

func NewOptionalApply 

func NewOptionalApply(outer, inner Operator, arg *Argument, paddedWidth int) *OptionalApply

NewOptionalApply creates an OptionalApply operator.

  • outer is the left (driving) plan.
  • inner is the right (sub) plan whose leftmost leaf is the provided arg.
  • arg is the Argument node at the inner leaf; OptionalApply seeds it before each inner Init call.
  • paddedWidth is the total width of an output row, i.e. outerWidth plus the number of columns the inner pipeline introduces. When the inner pipeline emits zero rows for an outer, OptionalApply emits a row of this width whose first outerWidth columns are the outer row and whose trailing columns are expr.Null.

OptionalApply takes ownership of both plans.

func (*OptionalApply) Close 

func (op *OptionalApply) Close() error

Close releases resources and closes both the outer and inner plans.

func (*OptionalApply) Init 

func (op *OptionalApply) Init(ctx context.Context) error

Init initialises both the outer plan and stores ctx for subsequent Next calls.

func (*OptionalApply) Next 

func (op *OptionalApply) Next(out *Row) (bool, error)

Next emits the next output row. The semantics are:

  • For each outer row, drain the inner pipeline.
  • If the inner pipeline emits ≥1 row, those rows are forwarded verbatim.
  • If the inner pipeline emits 0 rows, a single NULL-extended row is emitted consisting of the outer columns followed by expr.Null for every inner column the pipeline would have introduced.

type OptionalExpand 

type OptionalExpand struct {
	// contains filtered or unexported fields
}

OptionalExpand is a Volcano pipeline operator that performs a single-hop expansion and emits a NULL-extended row when no edges match for an input node.

OptionalExpand is NOT safe for concurrent use.

func NewOptionalExpand 

func NewOptionalExpand(input Operator, fwd, rev csrAdjacency, cfg ExpandConfig) *OptionalExpand

NewOptionalExpand creates an OptionalExpand operator.

  • input is the upstream operator supplying node IDs.
  • fwd is the forward CSR adjacency.
  • rev is the reverse CSR adjacency (required for DirIn/DirBoth).
  • cfg is the Expand configuration (direction, edge-type filter, inputCol).

The NULL-extension row uses the same column layout as Expand: inputRow... || srcID || Null(edgeID) || Null(dstID).

func (*OptionalExpand) Close 

func (op *OptionalExpand) Close() error

Close closes the input and child operators.

func (*OptionalExpand) Init 

func (op *OptionalExpand) Init(ctx context.Context) error

Init initialises the operator.

func (*OptionalExpand) Next 

func (op *OptionalExpand) Next(out *Row) (bool, error)

Next emits the next row. For each input row:

  • It feeds the row into the inner Expand one hop at a time.
  • If Expand emits ≥1 row, those rows are forwarded as-is.
  • If Expand emits 0 rows for an input node, a NULL-extended row is emitted.

type ParallelCountScan added in v0.6.0 

type ParallelCountScan struct {
	// contains filtered or unexported fields
}

ParallelCountScan is a Volcano leaf operator that computes a group-by-less count over a full node scan using per-worker partial counters combined once at the end. It emits exactly one row with a single expr.IntegerValue column carrying the total live-node count.

ParallelCountScan is NOT safe for concurrent use.

func NewParallelCountScan added in v0.6.0 

func NewParallelCountScan(g nodeWalker, morselSize int, gov *ParallelGovernor) *ParallelCountScan

NewParallelCountScan creates a ParallelCountScan over g. morselSize controls the chunk size per worker; pass 0 to use DefaultMorselSize. gov is the engine-shared adaptive worker-budget governor (nil = unbounded GOMAXPROCS).

func (*ParallelCountScan) Close added in v0.6.0 

func (op *ParallelCountScan) Close() error

Close cancels any still-running workers and joins them. It is idempotent and safe whether or not Next was ever called: wg.Wait returns immediately once the workers have drained, and cancel unblocks a worker stalled on ctx.

func (*ParallelCountScan) Init added in v0.6.0 

func (op *ParallelCountScan) Init(ctx context.Context) error

Init collects all node IDs, partitions them into morsels, and launches worker goroutines that each accumulate a private count. The combine is deferred to the first Next call so every worker is joined on the Next goroutine, inside the engine's visibility barrier.

func (*ParallelCountScan) Next added in v0.6.0 

func (op *ParallelCountScan) Next(out *Row) (bool, error)

Next emits the single aggregated row on its first call. It joins every worker synchronously (wg.Wait) on the calling goroutine, then sums the per-worker partials. Subsequent calls report end-of-stream.

type ParallelGovernor added in v0.6.0 

type ParallelGovernor struct {
	// contains filtered or unexported fields
}

ParallelGovernor bounds intra-query parallelism across the concurrently executing queries that share one Engine. A morsel-parallel leaf registers on Init via ParallelGovernor.Enter (which returns its worker budget) and deregisters on Close via ParallelGovernor.Leave.

The budget is GOMAXPROCS divided by the number of parallel leaves currently in flight (including the caller), clamped to [1, morsels]:

  • a single parallel query in flight gets the full GOMAXPROCS workers — no change from the prior unconditional behaviour, so single-query latency is unaffected;
  • N concurrent parallel queries each get ~GOMAXPROCS/N workers, so the aggregate worker count stays near GOMAXPROCS and the scheduler/cache thrash of N×GOMAXPROCS goroutines is avoided.

The inflight count is sampled once per Enter and races at the edges (an early query may briefly see a low count and grab a large budget); this is a deliberate approximation — it need only prevent the sustained N×GOMAXPROCS explosion, not divide the machine exactly. Worker count never affects results, only timing, so the governor has no ACID or openCypher-TCK impact.

A nil *ParallelGovernor is valid and means "unbounded" — every leaf gets the full GOMAXPROCS budget and no count is tracked. This preserves the prior behaviour for any caller that constructs an operator without a governor (the public BuildPlan path and the operator unit tests).

ParallelGovernor is safe for concurrent use by any number of goroutines.

func (*ParallelGovernor) Enter added in v0.6.0 

func (g *ParallelGovernor) Enter(morsels int) int

Enter registers one parallel leaf and returns its worker budget, an integer in [1, morsels]. It MUST be paired with exactly one ParallelGovernor.Leave (deferred in the operator's Close). On a nil governor it returns the full GOMAXPROCS budget (clamped to morsels) without tracking anything.

morsels is the number of work units the leaf has to distribute; the budget is never larger than that (more workers than morsels would leave workers idle).

func (*ParallelGovernor) Leave added in v0.6.0 

func (g *ParallelGovernor) Leave()

Leave deregisters a parallel leaf previously registered via ParallelGovernor.Enter. It is nil-safe and idempotency is the caller's responsibility (call it exactly once per successful Enter).

type ParallelScan 

type ParallelScan struct {
	// contains filtered or unexported fields
}

ParallelScan is a Volcano leaf operator that partitions the full node scan into fixed-size morsels and executes them concurrently on up to GOMAXPROCS worker goroutines.

ParallelScan is NOT safe for concurrent use.

func NewParallelScan 

func NewParallelScan(g nodeWalker, morselSize int) *ParallelScan

NewParallelScan creates a ParallelScan over g. morselSize controls the chunk size per worker; pass 0 to use DefaultMorselSize.

func (*ParallelScan) Close 

func (op *ParallelScan) Close() error

Close cancels workers and waits for them — and the closer goroutine — to exit. Joining closerWG (not just wg) guarantees no goroutine outlives Close, even when the caller never drained outCh (#1795).

func (*ParallelScan) Init 

func (op *ParallelScan) Init(ctx context.Context) error

Init collects all node IDs, partitions them into morsels, and launches worker goroutines. The output channel is populated lazily by the workers.

func (*ParallelScan) Next 

func (op *ParallelScan) Next(out *Row) (bool, error)

Next reads the next NodeID from the output channel.

type ParallelScanProject added in v0.6.0 

type ParallelScanProject struct {
	// contains filtered or unexported fields
}

ParallelScanProject is a Volcano leaf operator that partitions a full-node scan into morsels, runs an independent fused scan→filter→project sub-plan per morsel on up to GOMAXPROCS workers, and emits the concatenated per-worker result rows. The output schema is the projection's output schema (set by the factory's Project), so this operator's rows are ready for the engine's final column passthrough.

ParallelScanProject is NOT safe for concurrent use.

func NewParallelScanProject added in v0.6.0 

func NewParallelScanProject(g nodeWalker, factory SubplanFactory, morselSize int, gov *ParallelGovernor) *ParallelScanProject

NewParallelScanProject creates a ParallelScanProject over g whose per-worker fused sub-plans are built by factory. morselSize controls the chunk size per worker; pass 0 to use DefaultMorselSize.

func (*ParallelScanProject) Close added in v0.6.0 

func (op *ParallelScanProject) Close() error

Close cancels any still-running workers and joins them. It is idempotent and safe whether or not Next was ever called: wg.Wait returns immediately once the workers have drained, and cancel unblocks a worker stalled on ctx or inside a sub-plan's Next.

func (*ParallelScanProject) Init added in v0.6.0 

func (op *ParallelScanProject) Init(ctx context.Context) error

Init collects all node IDs, partitions them into morsels, builds one independent sub-plan per worker on the calling goroutine, and launches the workers. Each worker drives its sub-plan over the morsels it dequeues and accumulates deep-copied result rows into its private buffer. The join and combine are deferred to the first Next call so every worker is joined on the Next goroutine, inside the engine's visibility barrier.

func (*ParallelScanProject) Next added in v0.6.0 

func (op *ParallelScanProject) Next(out *Row) (bool, error)

Next streams the concatenated per-worker result rows. The first call joins every worker synchronously (wg.Wait) on the calling goroutine, surfaces the first worker error if any, then concatenates the per-worker buffers. Subsequent calls advance the cursor through the concatenation.

func (*ParallelScanProject) WithResultBudget added in v0.7.0 

func (op *ParallelScanProject) WithResultBudget(maxRows, maxBytes int64, estimateRow func(Row) int64) *ParallelScanProject

WithResultBudget threads the engine's per-query result-memory budget into the operator so the morsel workers stop accumulating once the fleet-wide total exceeds maxRows or maxBytes, bounding peak memory on the parallel path (#1830). maxRows/maxBytes of 0 leave that dimension unbounded (the engine convention); estimateRow is the engine's coarse per-row byte estimate and may be nil to disable byte-budget enforcement. It returns op for chaining and must be called before Init. When neither bound is set the operator behaves exactly as before (full materialisation), so the result multiset is unchanged under budget.

type ProcedureCallOp 

type ProcedureCallOp struct {
	// contains filtered or unexported fields
}

ProcedureCallOp invokes a registered procedure and emits its result rows.

ProcedureCallOp is NOT safe for concurrent use.

func NewProcedureCallOp 

func NewProcedureCallOp(
	namespace []string,
	name string,
	argExprs []func(Row) (expr.Value, error),
	yieldVars []string,
	child Operator,
	reg *procs.Registry,
) *ProcedureCallOp

NewProcedureCallOp creates a ProcedureCallOp.

namespace and name identify the procedure. argExprs evaluate procedure arguments against the current driver row. yieldVars names the output columns. child is the driving subplan; pass nil for a standalone CALL. reg is the procedure registry used for lookup at runtime.

func (*ProcedureCallOp) Close 

func (op *ProcedureCallOp) Close() error

Close releases resources and closes the child operator.

func (*ProcedureCallOp) Init 

func (op *ProcedureCallOp) Init(ctx context.Context) error

Init resets internal state and initialises the child if present.

func (*ProcedureCallOp) Next 

func (op *ProcedureCallOp) Next(out *Row) (bool, error)

Next advances the operator by one row.

It draws driving rows from the child (or a synthetic empty row when child is nil), invokes the procedure for each, buffers all result rows, and emits them one at a time.

Void procedure semantics. A procedure declared with no output columns (len(yieldVars) == 0) is treated as a side-effect-only step. When invoked in-query (op.child != nil) it must NOT consume the driver row; instead each driver row is emitted unchanged once the impl has run, preserving the upstream variable bindings for any downstream RETURN. Standalone CALL (op.child == nil) emits nothing.

type Project 

type Project struct {
	// contains filtered or unexported fields
}

Project is a Volcano pipeline operator that applies a list of ProjectionItem expressions to each input row, producing an output row with one column per item.

Project is NOT safe for concurrent use.

func NewProject 

func NewProject(child Operator, items []ProjectionItem) (*Project, error)

NewProject creates a Project operator. items defines the output schema; each item's Eval function is applied to each input row. An empty items slice is legal (e.g. `WITH *` over a pattern that binds no variables); the resulting operator forwards an empty Row for every input row.

func (*Project) Close 

func (op *Project) Close() error

Close releases resources and closes the child operator.

func (*Project) Columns 

func (op *Project) Columns() []string

Columns returns the ordered list of output column aliases.

func (*Project) Init 

func (op *Project) Init(ctx context.Context) error

Init initialises the operator and its child.

func (*Project) Next 

func (op *Project) Next(out *Row) (bool, error)

Next evaluates each projection item against the next input row and writes the result into out. Returns (true, nil) when a projected row is available, (false, nil) at end-of-stream, or (false, err) on evaluation or child error.

func (*Project) WithRowByteBudget added in v0.7.0 

func (op *Project) WithRowByteBudget(maxRowBytes int64, estimateValue func(expr.Value) int64) *Project

WithRowByteBudget bounds the estimated size of a single assembled output row by maxRowBytes, using estimateValue for the per-column estimate. It is enforced INCREMENTALLY inside Next — after each column is evaluated, before the next — so a projection of several large columns (e.g. RETURN range(1,N), range(1,N), …) is rejected before the whole row is materialised, bounding the transient peak to maxRowBytes plus one column regardless of the column count. It complements the drain's aggregate per-result byte budget, which is a retention guard on the SUM of already-built rows and therefore fires only after Next has assembled the row; this per-row guard moves the same accounting earlier and makes it per-column so construction cannot OOM (#1852). A non-positive maxRowBytes or nil estimateValue leaves the guard disabled (behaviour-preserving). Returns op for chaining; call before Init.

type ProjectionItem 

type ProjectionItem struct {
	// Alias is the output column name (e.g. "n", "count(n)", "x").
	Alias string
	// Eval evaluates the item expression against the current input row and
	// returns the projected value.
	Eval func(Row) (expr.Value, error)
}

ProjectionItem describes a single column in a projection. Eval is evaluated against the input row; Alias names the resulting output column.

type PropEntry 

type PropEntry struct {
	Key   string
	Value lpg.PropertyValue
}

PropEntry is an exported key/value pair for use by external plan builders (api.go) that construct dynamic property evaluators. It mirrors propLiteral but carries exported fields so the physical builder can return values from a PropsEvalFn without requiring propLiteral to be exported.

type PropsEvalFn 

type PropsEvalFn func(row Row) ([]PropEntry, error)

PropsEvalFn is a per-row property evaluator closure. It receives the current row and returns a slice of (key, value) pairs produced by evaluating the property-map AST expressions against the row's bound variables. Any entry whose evaluation yields Null is omitted (openCypher: assigning null to a property is a no-op on a fresh node).

It returns a non-nil error to fail-stop the statement — a runtime evaluation error, or a value that is not a valid property (a map or nested collection, InvalidPropertyType) — rather than silently dropping the entry (audit 2026-07-13 security F3 / fail-stop mandate).

The closure is constructed once by the physical plan builder and captures the schema, function registry, and query parameters.

type RangeBound 

type RangeBound struct {
	// Value is the bound's expr.Value.  Nil means unbounded (use the
	// minimum or maximum representable value for the index type).
	Value expr.Value
	// Include records the caller's intended inclusivity (≤ / ≥ vs < / >). It is
	// metadata only: NodeByIndexRangeScan always emits the inclusive [lo, hi]
	// superset and relies on a residual predicate Filter for exact open/closed
	// semantics (see the NodeByIndexRangeScan type doc, #F-EXEC1).
	Include bool
}

RangeBound carries one endpoint of a range predicate.

type Record 

type Record map[string]interface{}

Record is a single result row, accessed by column name. The underlying map is owned by the ResultSet; callers must copy values they need to retain beyond the next ResultSet.Next call.

type RelCols 

type RelCols struct {
	SrcCol  int
	DstCol  int
	EdgeCol int
}

RelCols carries the raw column indices that the Expand operator places for a relationship variable. SetProperty and RemoveProperty use it to reconstruct the (src, dst) endpoint keys when the bound entity is a relationship rather than a node.

SrcCol and DstCol hold the endpoint NodeIDs as IntegerValue. EdgeCol holds the forward-CSR edge-position counter (= schema[entityVar]); it is used to resolve the bound parallel instance's stable handle via GraphMutator.EdgeHandleAtPosition so a relationship SET/REMOVE maintains the per-instance by-handle property store (#1686). EdgeCol may be 0 only when it genuinely is column 0; callers distinguish "no handle" by the resolver returning 0, not by the column index.

type RelEndpointFn 

type RelEndpointFn func(row Row) (uint64, uint64, bool)

RelEndpointFn returns the (srcID, dstID) endpoints for an edge that the schema-direct path is about to delete. Used when the bare-variable target carries an IntegerValue edge id (the in-pipeline encoding emitted by Expand) so DeleteNode can dispatch to the edge-removal branch without misinterpreting the id as a node id.

type RemoveLabels 

type RemoveLabels struct {
	// contains filtered or unexported fields
}

RemoveLabels removes one or more labels from an already-bound node per input row.

RemoveLabels is NOT safe for concurrent use.

func NewRemoveLabels 

func NewRemoveLabels(
	nodeVar string,
	labels []string,
	schema map[string]int,
	child Operator,
	mutator GraphMutator,
) *RemoveLabels

NewRemoveLabels creates a RemoveLabels operator.

func (*RemoveLabels) Close 

func (op *RemoveLabels) Close() error

Close closes the child operator.

func (*RemoveLabels) Init 

func (op *RemoveLabels) Init(ctx context.Context) error

Init initialises the operator and its child.

func (*RemoveLabels) Next 

func (op *RemoveLabels) Next(out *Row) (bool, error)

Next pulls one row from the child and removes the specified labels.

type RemoveProperty 

type RemoveProperty struct {
	// contains filtered or unexported fields
}

RemoveProperty removes a single named property from an already-bound node or relationship per input row. For relationships, call WithRelCols to supply the endpoint column indices.

RemoveProperty is NOT safe for concurrent use.

func NewRemoveProperty 

func NewRemoveProperty(
	entityVar, propertyKey string,
	schema map[string]int,
	child Operator,
	mutator GraphMutator,
) *RemoveProperty

NewRemoveProperty creates a RemoveProperty operator.

func (*RemoveProperty) Close 

func (op *RemoveProperty) Close() error

Close closes the child operator.

func (*RemoveProperty) Init 

func (op *RemoveProperty) Init(ctx context.Context) error

Init initialises the operator and its child.

func (*RemoveProperty) Next 

func (op *RemoveProperty) Next(out *Row) (bool, error)

Next pulls one row from the child and removes the specified property.

func (*RemoveProperty) WithConstraintRegistry added in v0.3.0 

func (op *RemoveProperty) WithConstraintRegistry(reg *ConstraintRegistry) *RemoveProperty

WithConstraintRegistry attaches a ConstraintRegistry so RemoveProperty releases unique-constraint value reservations when a node property is removed. Returns op for chaining.

func (*RemoveProperty) WithRelCols 

func (op *RemoveProperty) WithRelCols(rc RelCols) *RemoveProperty

WithRelCols marks entityVar as a relationship variable and records the row columns that hold the src and dst NodeIDs. Must be called before the first Next invocation. Returns op for chaining.

type Result 

type Result interface {
	// Next advances to the next result row. It returns true if a row is
	// available, false at end-of-stream or on error. After Next returns false,
	// callers should check Err.
	Next() bool

	// Record returns the current row as a [Record] (column name → value).
	// Record must not be called before the first successful Next or after Next
	// returns false.
	Record() Record

	// Err returns the first error encountered during iteration, or nil.
	Err() error

	// Columns returns the ordered list of column names for this result set.
	// The slice is stable across calls and is not modified after construction.
	Columns() []string

	// Close releases all resources held by this result set, including the
	// underlying operator tree. It must be called exactly once.
	Close() error
}

Result is a forward-only, streaming iterator over query result rows.

Lifecycle

  1. Call [Next] in a loop until it returns false.
  2. After the loop, check [Err] for any error that terminated iteration.
  3. Call [Close] exactly once to release resources.

Concurrency

Result implementations are NOT safe for concurrent use.

type ResultSet 

type ResultSet struct {
	// contains filtered or unexported fields
}

ResultSet is the concrete implementation of Result returned by Run.

ResultSet is NOT safe for concurrent use.

func Run 

func Run(ctx context.Context, plan Operator, cols []string) *ResultSet

Run initialises plan, stores the column names, and returns a ResultSet ready for iteration. The caller drives iteration via ResultSet.Next and must call ResultSet.Close when done.

Run does not pull any rows; all work happens lazily in ResultSet.Next. The Record map is allocated lazily on the first ResultSet.Record / ResultSet.TakeRecord call (#1499 follow-up) and then reused across every subsequent Next call. A consumer that drains rows positionally via ResultSet.Row — the materialisation path and the Bolt PULL path — never triggers the allocation at all, which keeps small-result queries off the heap.

func (*ResultSet) Close 

func (rs *ResultSet) Close() error

Close releases all resources held by the ResultSet, including the underlying operator tree. It must be called exactly once. Calling Close after a previous Close is a no-op.

func (*ResultSet) Columns 

func (rs *ResultSet) Columns() []string

Columns returns the ordered list of column names. The slice is never nil and is stable for the lifetime of the ResultSet.

func (*ResultSet) Err 

func (rs *ResultSet) Err() error

Err returns the first error encountered during iteration, or nil.

func (*ResultSet) Next 

func (rs *ResultSet) Next() bool

Next advances to the next result row. It returns true when a row is available (accessible via Record), and false at end-of-stream or on error.

func (*ResultSet) Record 

func (rs *ResultSet) Record() Record

Record returns the current row as a map. Must only be called after a successful Next.

The returned map is owned by the ResultSet and reused by the next Next call; callers that need to retain a row beyond the next Next must copy it (or use ResultSet.TakeRecord). The map is built lazily on the first Record call for the current row, so a caller that consumes rows positionally via ResultSet.Row never pays for it.

func (*ResultSet) Row added in v0.3.1 

func (rs *ResultSet) Row() Row

Row returns the current row as a positional slice of values whose indices correspond to ResultSet.Columns. The slice is owned by the operator tree and is reused on the next ResultSet.Next call; callers that retain values beyond the next Next must copy them. This is the allocation-free accessor: unlike ResultSet.Record it never builds a map. Must only be called after a successful Next.

func (*ResultSet) RowCountHint added in v0.6.0 

func (rs *ResultSet) RowCountHint() (n int, ok bool)

RowCountHint reports a best-effort upper bound on the number of rows this ResultSet will yield, valid after Run (which calls Init). It returns ok=false when the plan exposes no sound upper bound. Callers use it purely to presize buffers; the value is never an exact count and must not be used to decide how many rows exist. See [rowCountHinter].

func (*ResultSet) TakeRecord 

func (rs *ResultSet) TakeRecord() Record

TakeRecord returns the current row and transfers ownership of its backing map to the caller, installing a fresh map for subsequent Next calls. Unlike ResultSet.Record — whose result is reused on the next Next — the map returned here is safe to retain. The materialisation path uses this to drain rows under the transaction-visibility barrier without the extra per-row copy that re-hashing every column into a new map would cost. Must only be called after a successful Next.

type RollUpApply 

type RollUpApply struct {
	// contains filtered or unexported fields
}

RollUpApply is a Volcano pipeline operator that performs pattern-comprehension execution: for each outer row, it drains the entire inner sub-plan into a expr.ListValue and emits (outerRow... || listValue) as a single output row.

RollUpApply is NOT safe for concurrent use.

func NewRollUpApply 

func NewRollUpApply(outer, inner Operator, arg *Argument, listEval func(Row) (expr.Value, error)) *RollUpApply

NewRollUpApply creates a RollUpApply operator with the default per-list element budget (funcs.DefaultMaxCollectItems).

  • outer is the driving (left) plan.
  • inner is the correlated (right) sub-plan whose leaf is arg.
  • arg is the Argument node seeded with each outer row before inner Init.
  • listEval, when non-nil, is called for each inner row to extract the value to collect into the list. When nil, the first column of each inner row is collected.

func NewRollUpApplyN added in v0.2.0 

func NewRollUpApplyN(outer, inner Operator, arg *Argument, listEval func(Row) (expr.Value, error), maxItems int) *RollUpApply

NewRollUpApplyN is NewRollUpApply with an explicit per-list element budget. maxItems uses the EngineOptions.MaxCollectItems encoding so the cap is consistent with the buffering aggregators (#1294): 0 selects funcs.DefaultMaxCollectItems, a negative value disables the cap entirely, and a positive value is used verbatim. The encoding is resolved once here so the drain loop compares against a single non-negative ceiling.

func (*RollUpApply) Close 

func (op *RollUpApply) Close() error

Close closes both the outer and inner plans.

func (*RollUpApply) Init 

func (op *RollUpApply) Init(ctx context.Context) error

Init initialises the outer plan.

func (*RollUpApply) Next 

func (op *RollUpApply) Next(out *Row) (bool, error)

Next emits one output row per outer row. For each outer row, the entire inner plan is drained into a ListValue appended as a new column.

type Row 

type Row []expr.Value

Row is a single tuple in the pipeline: a slice of expr.Value whose positions correspond to the operator's output schema. The slice is owned by the RowSlab that allocated it; callers must not retain it beyond the slab's lifetime.

func Drain 

func Drain(ctx context.Context, op Operator) ([]Row, error)

Drain initialises op, pulls every row from the pipeline, and returns the collected rows as a []Row. Close is always called before Drain returns, regardless of whether an error occurred.

Cancellation: Drain honours ctx.Done() via the per-Next check inside each operator. If ctx is cancelled, Drain returns the partial result set and the context error.

The returned rows are independent copies: each element is a snapshot of the Row written by the operator at that iteration. Callers own the returned slice.

Example

ExampleDrain assembles the equivalent of `UNWIND [10, 20, 30] AS x RETURN x LIMIT 2` as a hand-built operator tree and drains it. SingleRow seeds one empty row, Unwind expands the literal list, and Limit caps the output. 

package main

import (
	"context"
	"fmt"

	"github.com/FlavioCFOliveira/GoGraph/cypher/exec"
	"github.com/FlavioCFOliveira/GoGraph/cypher/expr"
)

func main() {
	// SingleRow emits exactly one empty row to drive the pipeline.
	src := exec.NewSingleRowOperator()

	// Unwind expands a fixed list; listFn ignores the (empty) input row.
	unwind, err := exec.NewUnwind(src, func(exec.Row) (expr.ListValue, error) {
		return expr.ListValue{
			expr.IntegerValue(10),
			expr.IntegerValue(20),
			expr.IntegerValue(30),
		}, nil
	})
	if err != nil {
		fmt.Println("NewUnwind:", err)
		return
	}

	// Limit passes at most two rows downstream.
	limit, err := exec.NewLimit(unwind, 2)
	if err != nil {
		fmt.Println("NewLimit:", err)
		return
	}

	// Drain runs the pipeline and always closes it before returning.
	rows, err := exec.Drain(context.Background(), limit)
	if err != nil {
		fmt.Println("Drain:", err)
		return
	}

	fmt.Println("rows:", len(rows))
	for _, r := range rows {
		fmt.Println("x =", r[0])
	}
}

Output:
rows: 2
x = 10
x = 20

type RowSlab 

type RowSlab struct {
	// contains filtered or unexported fields
}

RowSlab is a bounded arena of pre-allocated rows. It eliminates per-row heap allocations by backing all rows in a single flat slice. Each call to [Alloc] hands out a sub-slice at zero GC cost after the initial backing allocation.

RowSlab is NOT safe for concurrent use; each pipeline stage owns its own instance, typically obtained from SlabPool.

Lifecycle

  1. Obtain a slab from SlabPool.Get (or call NewRowSlab).
  2. Call [Alloc] for each row needed in the current batch.
  3. When the batch is fully processed, call [Reset] and return the slab to SlabPool.Put.

[Reset] zeroes the column values in every allocated row so that no expr.Value reference is retained past the batch boundary (preventing GC nepotism between batches).

func NewRowSlab 

func NewRowSlab(width, capacity int) *RowSlab

NewRowSlab creates a RowSlab with the given column count and row capacity. width is the number of expr.Value slots pre-allocated per row; pass 0 for variable-width rows (callers supply their own slice to [AllocRaw]). capacity must be ≥ 1; DefaultSlabCapacity is a reasonable default.

func (*RowSlab) Alloc 

func (s *RowSlab) Alloc() (Row, error)

Alloc returns the next available pre-allocated row in the slab. It returns ErrSlabOverflow if the slab is exhausted. The row width matches the width passed to NewRowSlab; for variable-width slabs (width=0) use [AllocRaw].

func (*RowSlab) AllocRaw 

func (s *RowSlab) AllocRaw() (int, error)

AllocRaw returns the next row slot for variable-width slabs (width=0). The caller is responsible for setting the returned row to a correctly-sized slice before use. For fixed-width slabs, use [Alloc].

func (*RowSlab) Cap 

func (s *RowSlab) Cap() int

Cap returns the maximum number of rows this slab can hold.

func (*RowSlab) GetRow 

func (s *RowSlab) GetRow(idx int) Row

GetRow returns the row at index idx. Panics if idx is out of bounds.

func (*RowSlab) Len 

func (s *RowSlab) Len() int

Len returns the number of rows currently allocated.

func (*RowSlab) Reset 

func (s *RowSlab) Reset()

Reset resets the slab for reuse. It zeroes every value slot in each allocated row to release references held by the GC, then resets the allocation counter to zero. The backing memory is retained.

func (*RowSlab) SetRow 

func (s *RowSlab) SetRow(idx int, r Row)

SetRow stores row r at index idx. Panics if idx is out of bounds. Used with variable-width slabs after [AllocRaw].

type SemiApply 

type SemiApply struct {
	// contains filtered or unexported fields
}

SemiApply emits each outer row for which the inner sub-plan produces at least one row. The inner plan is closed after the first match (short-circuit).

SemiApply is NOT safe for concurrent use.

func NewSemiApply 

func NewSemiApply(outer, inner Operator, arg *Argument) *SemiApply

NewSemiApply creates a SemiApply operator.

  • outer is the driving (left) plan.
  • inner is the correlated (right) sub-plan whose leaf is arg.
  • arg is the Argument node seeded with each outer row before inner Init.

func (*SemiApply) Close 

func (op *SemiApply) Close() error

Close closes the outer plan. The inner plan is already closed per-row inside Next; a redundant close here is safe because Close on a closed operator must be a no-op per the Operator contract.

func (*SemiApply) Init 

func (op *SemiApply) Init(ctx context.Context) error

Init initialises the outer plan.

func (*SemiApply) Next 

func (op *SemiApply) Next(out *Row) (bool, error)

Next advances to the next outer row for which the inner plan has ≥1 result.

type SetAllProperties 

type SetAllProperties struct {
	// contains filtered or unexported fields
}

SetAllProperties replaces or merges every property on an already-bound node or relationship per input row.

SetAllProperties is NOT safe for concurrent use.

func NewSetAllPropertiesFromEntity 

func NewSetAllPropertiesFromEntity(
	entityVar, sourceVar string,
	isReplace bool,
	schema map[string]int,
	child Operator,
	mutator GraphMutator,
) *SetAllProperties

NewSetAllPropertiesFromEntity creates a SetAllProperties operator copying every property from sourceVar (a bound node or relationship) to entityVar. isReplace selects `=` (true) vs `+=` (false) semantics.

func NewSetAllPropertiesFromMap 

func NewSetAllPropertiesFromMap(
	entityVar, mapLiteral string,
	isReplace bool,
	schema map[string]int,
	child Operator,
	mutator GraphMutator,
) (*SetAllProperties, error)

NewSetAllPropertiesFromMap creates a SetAllProperties operator writing every key/value pair from mapLiteral to entityVar. mapLiteral is the opaque literal-map string (e.g. `{a: 1, b: "x"}`) produced by the AST printer.

func NewSetAllPropertiesFromParam 

func NewSetAllPropertiesFromParam(
	entityVar, paramName string,
	isReplace bool,
	schema map[string]int,
	child Operator,
	mutator GraphMutator,
) *SetAllProperties

NewSetAllPropertiesFromParam creates a SetAllProperties operator writing every key/value pair from the named query parameter to entityVar. The parameter must resolve to a MapValue at exec time; non-map values are treated as a no-op.

func (*SetAllProperties) Close 

func (op *SetAllProperties) Close() error

Close closes the child operator.

func (*SetAllProperties) Init 

func (op *SetAllProperties) Init(ctx context.Context) error

Init initialises the operator and its child.

func (*SetAllProperties) Next 

func (op *SetAllProperties) Next(out *Row) (bool, error)

Next pulls one row from the child and applies the configured whole-entity mutation. The row is forwarded unchanged so downstream operators (e.g. ProduceResults) can read the affected entity.

func (*SetAllProperties) WithConstraints 

func (op *SetAllProperties) WithConstraints(reg *ConstraintRegistry, mgr *index.Manager) *SetAllProperties

WithConstraints attaches a ConstraintRegistry and index.Manager for pre-write enforcement. Both must be non-nil. Returns op for chaining.

func (*SetAllProperties) WithParams 

func (op *SetAllProperties) WithParams(params map[string]expr.Value) (*SetAllProperties, error)

WithParams attaches query parameters for $name substitution in the literal map and for parameter-sourced operators. Returns op for chaining.

func (*SetAllProperties) WithRelCols 

func (op *SetAllProperties) WithRelCols(rc RelCols) *SetAllProperties

WithRelCols marks entityVar as a relationship variable and records the row columns that hold the src and dst NodeIDs. Must be called before the first Next invocation. Returns op for chaining.

func (*SetAllProperties) WithSourceRelCols 

func (op *SetAllProperties) WithSourceRelCols(rc RelCols) *SetAllProperties

WithSourceRelCols marks sourceVar as a relationship variable and records the row columns that hold its src and dst NodeIDs. Must be called before the first Next invocation when SourceVar is a relationship. Returns op for chaining.

type SetLabels 

type SetLabels struct {
	// contains filtered or unexported fields
}

SetLabels adds one or more labels to an already-bound node per input row.

SetLabels is NOT safe for concurrent use.

func NewSetLabels 

func NewSetLabels(
	nodeVar string,
	labels []string,
	schema map[string]int,
	child Operator,
	mutator GraphMutator,
) *SetLabels

NewSetLabels creates a SetLabels operator.

func (*SetLabels) Close 

func (op *SetLabels) Close() error

Close closes the child operator.

func (*SetLabels) Init 

func (op *SetLabels) Init(ctx context.Context) error

Init initialises the operator and its child.

func (*SetLabels) Next 

func (op *SetLabels) Next(out *Row) (bool, error)

Next pulls one row from the child and adds the specified labels.

type SetProperty 

type SetProperty struct {
	// contains filtered or unexported fields
}

SetProperty sets or replaces properties on an already-bound node or relationship per input row. The entity is identified by entityVar. For nodes, the column value must be an IntegerValue-encoded NodeID or a NodeValue. For relationships, call WithRelCols to supply the endpoint column indices.

SetProperty is NOT safe for concurrent use.

func NewSetProperty 

func NewSetProperty(
	entityVar, propertyKey, valueExpr string,
	schema map[string]int,
	child Operator,
	mutator GraphMutator,
) (*SetProperty, error)

NewSetProperty creates a SetProperty operator.

entityVar is the variable name of the target node or relationship. propertyKey is the property key for single-property mode; pass empty for whole-entity mode. valueExpr is the opaque literal string from the IR. schema maps variable names to column indices. mutator is the graph write surface.

func (*SetProperty) Close 

func (op *SetProperty) Close() error

Close closes the child operator.

func (*SetProperty) Init 

func (op *SetProperty) Init(ctx context.Context) error

Init initialises the operator and its child.

func (*SetProperty) Next 

func (op *SetProperty) Next(out *Row) (bool, error)

Next pulls one row from the child and applies the property mutation.

func (*SetProperty) WithConstraints 

func (op *SetProperty) WithConstraints(reg *ConstraintRegistry, mgr *index.Manager) *SetProperty

WithConstraints attaches a ConstraintRegistry and index.Manager for pre-write enforcement. Both must be non-nil. Returns op for chaining.

func (*SetProperty) WithParams 

func (op *SetProperty) WithParams(params map[string]expr.Value) *SetProperty

WithParams attaches query parameters for $name substitution in value expressions. Returns op for chaining.

func (*SetProperty) WithRelCols 

func (op *SetProperty) WithRelCols(rc RelCols) *SetProperty

WithRelCols marks entityVar as a relationship variable and records the row columns that hold the src and dst NodeIDs. Must be called before the first Next invocation. Returns op for chaining.

func (*SetProperty) WithValueEvalFn 

func (op *SetProperty) WithValueEvalFn(fn ValueEvalFn) *SetProperty

WithValueEvalFn attaches a per-row evaluator for the SET RHS expression. The closure is invoked whenever the operator needs to compute the new property value; it takes priority over the literal-string parser path for single-property assignments.

type ShortestPath 

type ShortestPath struct {
	// contains filtered or unexported fields
}

ShortestPath is a Volcano pipeline operator that, for each input row, finds a single shortest path from srcCol to dstCol using BFS and emits one output row containing the flat alternating path list (or Null if unreachable).

ShortestPath is NOT safe for concurrent use.

func NewShortestPath 

func NewShortestPath(input Operator, fwd, rev csrAdjacency, dir Direction, srcCol, dstCol int) *ShortestPath

NewShortestPath creates a ShortestPath operator.

  • input is the upstream operator supplying (srcID, dstID) pairs.
  • fwd is the forward CSR adjacency.
  • rev is the reverse CSR (required for DirIn/DirBoth).
  • dir is the traversal direction.
  • srcCol / dstCol are the column indices in each input row for source and destination node IDs.

The returned operator has no type filter and no hop bounds; use ShortestPath.WithTypeFilter and ShortestPath.WithHopBounds to configure them.

func (*ShortestPath) Close 

func (op *ShortestPath) Close() error

Close closes the input operator.

func (*ShortestPath) Init 

func (op *ShortestPath) Init(ctx context.Context) error

Init initialises the operator.

func (*ShortestPath) Next 

func (op *ShortestPath) Next(out *Row) (bool, error)

Next emits one row per input row, containing the shortest path. When no path exists the row is emitted with a Null path (OPTIONAL MATCH) or dropped (MATCH) depending on the operator's optional flag.

func (*ShortestPath) WithHopBounds added in v0.6.0 

func (op *ShortestPath) WithHopBounds(minHops, maxHops int) *ShortestPath

WithHopBounds sets the accepted path-length window. minHops is the minimum length (a zero-length src==dst path is reported only when minHops == 0); maxHops caps the BFS depth ([shortestNoMaxHops] == 0 means unbounded). It returns op for chaining.

func (*ShortestPath) WithOptional added in v0.6.0 

func (op *ShortestPath) WithOptional(optional bool) *ShortestPath

WithOptional selects the OPTIONAL MATCH no-path behaviour (emit a Null-path row) when optional is true; the default (false) drops the row when no path exists, the MATCH behaviour. It returns op for chaining.

func (*ShortestPath) WithPathPredicate added in v0.6.0 

func (op *ShortestPath) WithPathPredicate(pred func(Row) (bool, error)) *ShortestPath

WithPathPredicate fuses a whole-path predicate onto the operator (#1786). The operator then returns the shortest path that SATISFIES pred (an exhaustive search), instead of the unconstrained shortest path. pred is called with the candidate's full output row. It returns op for chaining.

func (*ShortestPath) WithTypeFilter added in v0.6.0 

func (op *ShortestPath) WithTypeFilter(edgeType string, filter map[uint64]string) *ShortestPath

WithTypeFilter restricts traversal to edges whose forward position is present in filter. edgeType is the non-empty "a filter was requested" gate (typically the pattern's first declared relationship type). It returns op for chaining.

func (*ShortestPath) WithWorkBudget added in v0.7.0 

func (op *ShortestPath) WithWorkBudget(maxPerRow, maxTotal int) *ShortestPath

WithWorkBudget overrides the exhaustive path-predicate search's per-input-row and aggregate per-query edge-traversal caps (see [ShortestPath.maxEdgesTraversed]). A non-positive value leaves the corresponding default in place. It returns op for chaining and is primarily a testing seam; production uses the defaults.

type SingleRow 

type SingleRow struct {
	// contains filtered or unexported fields
}

SingleRow emits one empty row then signals exhaustion.

SingleRow is NOT safe for concurrent use.

func NewSingleRowOperator 

func NewSingleRowOperator() *SingleRow

NewSingleRowOperator returns a SingleRow operator.

func (*SingleRow) Close 

func (op *SingleRow) Close() error

Close is a no-op; SingleRow holds no resources.

func (*SingleRow) Init 

func (op *SingleRow) Init(ctx context.Context) error

Init resets the operator state.

func (*SingleRow) Next 

func (op *SingleRow) Next(out *Row) (bool, error)

Next emits one empty row on the first call and returns false on all subsequent calls.

type Skip 

type Skip struct {
	// contains filtered or unexported fields
}

Skip is a Volcano pipeline operator that discards the first n rows from its child operator and then forwards all remaining rows.

Skip is NOT safe for concurrent use.

func NewSkip 

func NewSkip(child Operator, n int64) (*Skip, error)

NewSkip creates a Skip operator that discards the first n rows from child. n must be ≥ 0; a skip of 0 is a no-op pass-through.

func (*Skip) Close 

func (op *Skip) Close() error

Close releases resources and closes the child operator.

func (*Skip) Init 

func (op *Skip) Init(ctx context.Context) error

Init initialises the operator and resets the skip counter.

func (*Skip) Next 

func (op *Skip) Next(out *Row) (bool, error)

Next discards rows until n have been skipped, then forwards subsequent rows.

type SlabPool 

type SlabPool struct {
	// contains filtered or unexported fields
}

SlabPool is a sync.Pool-backed pool of RowSlab instances with a fixed column width and capacity. Operators that process a high volume of rows should obtain slabs from a shared pool to reduce GC pressure.

SlabPool is safe for concurrent use.

func NewSlabPool 

func NewSlabPool(width, capacity int) *SlabPool

NewSlabPool creates a SlabPool that vends RowSlabs with the given column width and row capacity.

func (*SlabPool) Get 

func (sp *SlabPool) Get() *RowSlab

Get retrieves a reset RowSlab from the pool, or allocates a new one.

func (*SlabPool) Put 

func (sp *SlabPool) Put(s *RowSlab)

Put resets s and returns it to the pool.

type Sort 

type Sort struct {
	// contains filtered or unexported fields
}

Sort is a blocking Volcano operator that collects all rows from its child, sorts them by the specified SortKey sequence, and emits them in order.

Sort is NOT safe for concurrent use.

func NewSort 

func NewSort(child Operator, keys []SortKey, maxRows int) (*Sort, error)

NewSort creates a Sort operator.

  • child: the upstream operator to consume.
  • keys: ORDER BY specification. Must not be empty.
  • maxRows: upper bound on rows held in memory; pass 0 to use DefaultMaxSortRows.

func (*Sort) Close 

func (op *Sort) Close() error

Close closes the child operator and releases internal storage.

func (*Sort) Init 

func (op *Sort) Init(ctx context.Context) error

Init initialises the operator. The blocking collect+sort phase is deferred to the first Next call.

func (*Sort) Next 

func (op *Sort) Next(out *Row) (bool, error)

Next emits the next sorted row. On the first call it collects and sorts all rows from the child (pipeline breaker). Subsequent calls step through the sorted slice.

func (*Sort) WithByteBudget added in v0.7.0 

func (op *Sort) WithByteBudget(maxBytes int64, estimateRow func(Row) int64) *Sort

WithByteBudget bounds the estimated retained size of the buffered rows by maxBytes, using estimateRow for the per-row estimate. It complements the maxRows count cap so a few large-valued rows cannot exceed the engine's result-byte budget before the count cap fires (#1841). A non-positive maxBytes or nil estimateRow leaves the byte dimension disabled. Returns op for chaining and must be called before Init.

type SortKey 

type SortKey struct {
	// ColIdx is the zero-based index of the column within each Row.
	// Ignored when Eval is non-nil.
	ColIdx int
	// Ascending controls the sort direction. true = ASC, false = DESC.
	Ascending bool
	// Eval is an optional expression evaluator. When non-nil the sort key
	// value is obtained by calling Eval(row) rather than reading row[ColIdx].
	// This supports ORDER BY expressions that are not direct projection
	// output columns (e.g. ORDER BY n.age after RETURN n).
	Eval func(Row) (expr.Value, error)
}

SortKey describes a single ORDER BY column.

type StringHashIndex 

type StringHashIndex struct {
	// contains filtered or unexported fields
}

StringHashIndex adapts hash.Index[string] to the [hashLookup] interface. It accepts only expr.StringValue seek keys; other kinds return ErrIndexTypeMismatch.

func NewStringHashIndex 

func NewStringHashIndex(idx interface {
	LookupAppend(value string, dst []uint64) []uint64
}) *StringHashIndex

NewStringHashIndex constructs a StringHashIndex.

func (*StringHashIndex) LookupAppend added in v0.6.0 

func (h *StringHashIndex) LookupAppend(value expr.Value, dst []uint64) ([]uint64, error)

LookupAppend implements [hashLookup].

type StringRangeIndex 

type StringRangeIndex struct {
	// contains filtered or unexported fields
}

StringRangeIndex adapts btree.Index[string] to the [rangeLookup] interface. An unbounded lower bound is "" (the true minimum of the string order); an unbounded (or non-string) upper bound routes to the index's open-ended RangeFrom scan rather than a fixed sentinel — no fixed key is a true maximum for a variable-length string, so a sentinel cap would silently drop any key sorting above it (#F-CY1).

func NewStringRangeIndex 

func NewStringRangeIndex(idx interface {
	Range(lo, hi string) *roaring64.Bitmap
	RangeFrom(lo string) *roaring64.Bitmap
}) *StringRangeIndex

NewStringRangeIndex constructs a StringRangeIndex.

func (*StringRangeIndex) RangeBitmap 

func (r *StringRangeIndex) RangeBitmap(lo, hi expr.Value) *roaring64.Bitmap

RangeBitmap implements [rangeLookup]. An unbounded-above range (nil/NULL or a non-string upper bound) scans open-ended via RangeFrom so the bitmap is a genuine superset of every key >= lo (#F-CY1); a bounded upper uses the inclusive [lo, hi] Range.

type SubplanFactory added in v0.6.0 

type SubplanFactory func(morsel []graph.NodeID) (Operator, error)

SubplanFactory builds an independent physical sub-plan that scans exactly the NodeIDs in morsel and applies the fused Filter/Projection over them. Each call must return a fresh operator tree that shares NO mutable state with any other call's tree (the planner achieves this with a per-worker schema map and a per-worker buildOpts copy whose lazily-written fields are zeroed). The morsel slice is owned by the caller (ParallelScanProject) and is read-only for the lifetime of the returned operator; the factory must not retain or mutate it beyond feeding it to the morsel scan leaf.

The returned operator is driven Init → Next* → Close by exactly one worker goroutine. It must honour the standard Operator lifecycle.

type TargetEvalFn 

type TargetEvalFn func(row Row) (expr.Value, error)

TargetEvalFn evaluates a DELETE / DETACH DELETE target expression against the current input row and returns the resolved value. The exec operator inspects the value: NodeValue / IntegerValue selects the node by ID; RelationshipValue selects the relationship; null is a row-passthrough no-op (matches openCypher 9 §3.5.8).

type Top 

type Top struct {
	// contains filtered or unexported fields
}

Top is a blocking Volcano operator that emits the N smallest rows (per the given sort keys) from its child, using a bounded heap for O(M log N) memory and time.

Top is NOT safe for concurrent use.

func NewTop 

func NewTop(child Operator, keys []SortKey, n int) (*Top, error)

NewTop creates a Top operator.

  • child: the upstream operator to consume.
  • keys: ORDER BY specification. Must not be empty.
  • n: number of rows to return. Must be ≥ 0; n == 0 yields an empty result while still draining the child (ORDER BY … LIMIT 0, see #1801).

func (*Top) Close 

func (op *Top) Close() error

Close closes the child operator and releases internal storage.

func (*Top) Init 

func (op *Top) Init(ctx context.Context) error

Init initialises the operator. The blocking consume phase is deferred to the first Next call.

func (*Top) Next 

func (op *Top) Next(out *Row) (bool, error)

Next emits the next top-N row in sorted order. On the first call it consumes all rows from the child and finalises the heap.

type Union 

type Union struct {
	// contains filtered or unexported fields
}

Union emits the set-union of left and right: all rows from both sides with duplicates removed. It is implemented as UnionAll wrapped in a Distinct operator.

Schema mismatch detection is inherited from UnionAll.

Union is NOT safe for concurrent use.

func NewUnion 

func NewUnion(left, right Operator, maxDistinct int) *Union

NewUnion creates a Union operator that deduplicates the concatenation of left and right.

  • maxDistinct: upper bound on distinct rows; pass 0 to use DefaultMaxDistinct.

func (*Union) Close 

func (op *Union) Close() error

Close releases all resources.

func (*Union) Init 

func (op *Union) Init(ctx context.Context) error

Init initialises the operator.

func (*Union) Next 

func (op *Union) Next(out *Row) (bool, error)

Next emits the next unique row from the union.

type UnionAll 

type UnionAll struct {
	// contains filtered or unexported fields
}

UnionAll is a Volcano operator that concatenates the output of left and right children without deduplication. It validates that both sides produce rows of the same width (column count).

UnionAll is NOT safe for concurrent use.

func NewUnionAll 

func NewUnionAll(left, right Operator) *UnionAll

NewUnionAll creates a UnionAll operator that concatenates left then right.

func (*UnionAll) Close 

func (op *UnionAll) Close() error

Close closes both child operators. Both are always attempted; if both fail the errors are joined.

func (*UnionAll) Init 

func (op *UnionAll) Init(ctx context.Context) error

Init initialises both child operators.

func (*UnionAll) Next 

func (op *UnionAll) Next(out *Row) (bool, error)

Next emits rows from left until exhausted, then emits rows from right. Returns ErrSchemaMismatch if the first right-side row has a different column count than the first left-side row.

type Unwind 

type Unwind struct {
	// contains filtered or unexported fields
}

Unwind is a Volcano pipeline operator that implements the UNWIND clause. For each input row it evaluates a list expression and emits one output row per list element, appending the element value as a new column.

Unwind is NOT safe for concurrent use.

func NewUnwind 

func NewUnwind(child Operator, listFn UnwindListFn) (*Unwind, error)

NewUnwind creates an Unwind operator.

child provides the context rows. listFn is evaluated once per input row and must return the list to expand. The caller is responsible for appending the element column to the output row; Unwind handles that internally.

Both child and listFn are required: a nil argument returns the typed sentinel ErrUnwindNilChild or ErrUnwindNilListFn respectively, so callers can distinguish the cause via errors.Is. NewUnwind never panics.

func (*Unwind) Close 

func (op *Unwind) Close() error

Close releases resources and closes the child operator.

Close is idempotent within a single pipeline lifecycle: calling it more than once between two Init invocations returns nil from the second and later calls and does NOT propagate to op.child.Close again. The idempotency guard is reset by Init, so an Init→Close→Init→Close sequence still closes the child twice — once per cycle, as expected.

func (*Unwind) Init 

func (op *Unwind) Init(ctx context.Context) error

Init initialises the operator and its child. It clears all per-iteration state (curRow, curList, listIdx) and resets the idempotency guard (closed) so that Init is the exact dual of Close, allowing an operator instance to be safely re-Init'd after a previous Close.

func (*Unwind) Next 

func (op *Unwind) Next(out *Row) (bool, error)

Next advances to the next element. It pulls a new input row from the child whenever the current list is exhausted, then emits one row per element.

Returns (true, nil) when an output row was written to out, (false, nil) at end-of-stream, (false, err) on error.

type UnwindListFn 

type UnwindListFn func(row Row) (expr.ListValue, error)

UnwindListFn evaluates the list expression for one input row. It returns a expr.ListValue when the expression evaluates to a list, or nil/empty when there is nothing to expand.

type ValueEvalFn 

type ValueEvalFn func(row Row) (value lpg.PropertyValue, isNull bool, hasValue bool, err error)

ValueEvalFn evaluates a SET RHS expression against the current input row and returns the resulting property value plus a flag distinguishing the "no value produced" case from the "explicit null" case. The exec operator uses null/no-value semantics to either delete (null) or no-op (no value).

type VarLengthConfig 

type VarLengthConfig struct {
	// Direction to follow. Defaults to DirOut when zero.
	Direction Direction
	// EdgeType, when non-empty, restricts expansion to edges of this type.
	EdgeType string
	// EdgeTypeFilter maps absolute edge positions to type labels.
	EdgeTypeFilter map[uint64]string
	// InputCol is the column index in each input row that holds the source
	// NodeID. Defaults to 0.
	InputCol int
	// MinHops is the minimum path length (inclusive). Must be ≥ 0.
	MinHops int
	// MaxHops is the maximum path length (inclusive). Must be ≥ MinHops.
	// Use math.MaxInt for unbounded (not recommended without a safety cap).
	MaxHops int
	// MaxEdgesTraversed is the safety cap on edge traversals per input row.
	// Defaults to [defaultMaxEdgesTraversed] (1,000,000) when 0.
	MaxEdgesTraversed int
	// MaxTotalEdgesTraversed is the aggregate safety cap on edge traversals
	// across all input rows for the whole query — it is NOT reset per row, so it
	// bounds the M × (per-row cost) multiplication that the per-row cap alone
	// cannot (#1478). Defaults to [defaultMaxTotalEdgesTraversed] (100,000,000)
	// when 0.
	MaxTotalEdgesTraversed int
	// ExcludedRelCols lists column indices in the input row holding edge
	// identifiers (IntegerValue or RelationshipValue) that must not be
	// traversed inside this VLE step. Implements the openCypher
	// no-repeated-relationships rule across distinct rel patterns within
	// the same MATCH (e.g. `MATCH ()-[r:EDGE]-() MATCH (n)-[*0..1]-()-[r]
	// -()-[*0..1]-(m)` — the two variable-length steps must not reuse the
	// edge bound to `r`). The visited bitset is pre-populated with each
	// listed column's edge position at BFS seed time.
	ExcludedRelCols []int
}

VarLengthConfig carries configuration for NewVarLengthExpand.

type VarLengthExpand 

type VarLengthExpand struct {
	// contains filtered or unexported fields
}

VarLengthExpand is a Volcano pipeline operator that performs bounded BFS variable-length expansion.

VarLengthExpand is NOT safe for concurrent use.

func NewVarLengthExpand 

func NewVarLengthExpand(input Operator, fwd, rev csrAdjacency, cfg *VarLengthConfig) *VarLengthExpand

NewVarLengthExpand creates a VarLengthExpand operator. cfg is read-only and taken by pointer to avoid copying the configuration struct on this hot path.

func (*VarLengthExpand) Close 

func (op *VarLengthExpand) Close() error

Close releases resources.

func (*VarLengthExpand) Init 

func (op *VarLengthExpand) Init(ctx context.Context) error

Init initialises the operator.

func (*VarLengthExpand) Next 

func (op *VarLengthExpand) Next(out *Row) (bool, error)

Next emits the next (inputRow... || pathEdgesAsListValue || dstNodeID) row. The path is encoded as a expr.ListValue of edge positions (IntegerValues), followed by the destination node ID as an IntegerValue.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.