fabric

package
v1.2.2 Latest Latest
Warning

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

 Go to latest Published: Aug 6, 2026 License: MIT Imports: 14 Imported by: 0

Documentation

Overview

Package fabric implements a Neo4j Fabric-compatible distributed query layer.

It provides the Fragment plan tree, Catalog graph registry, FabricPlanner for decomposing Cypher queries at USE-clause boundaries, FabricExecutor for dispatching fragments to local or remote constituents, and FabricTransaction for coordinating per-shard sub-transactions with one-write-shard enforcement.

This package mirrors the architecture of org.neo4j.fabric in the Neo4j source, mapped to Go idioms and wired into NornicDB's existing storage and multidb layers.

Index

Constants

This section is empty.

Variables

View Source 
var ErrSecondWriteShard = fmt.Errorf("Neo.ClientError.Transaction.ForbiddenDueToTransactionType: " +
	"Writing to more than one database per transaction is not allowed")

ErrSecondWriteShard is returned when a distributed transaction attempts to write to a second shard. Neo4j Fabric enforces a many-read/one-write constraint per transaction — this error matches that contract with a stable code and message.

Functions

func RecordBindingsFromContext 

func RecordBindingsFromContext(ctx context.Context) (map[string]interface{}, bool)

RecordBindingsFromContext returns correlated outer-row bindings if present.

func WithFabricTransaction 

func WithFabricTransaction(ctx context.Context, tx *FabricTransaction) context.Context

WithFabricTransaction returns a context carrying the active Fabric transaction.

func WithHotPathTrace 

func WithHotPathTrace(ctx context.Context, trace *HotPathTrace) context.Context

WithHotPathTrace attaches a mutable trace object to context.

func WithRecordBindings 

func WithRecordBindings(ctx context.Context, bindings map[string]interface{}) context.Context

WithRecordBindings stores correlated outer-row bindings for Fabric APPLY execution.

func WithSubTransaction 

func WithSubTransaction(ctx context.Context, sub *SubTransaction) context.Context

WithSubTransaction returns a context carrying the active fabric sub-transaction.

Types

type Catalog 

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

Catalog is the registry of all known graphs and their locations. It mirrors Neo4j's Catalog.scala and is populated from multidb.DatabaseManager.

Graph names follow the pattern:

  • "dbname" — a standard or composite database
  • "composite.alias" — a constituent within a composite database

Thread-safe: all operations are protected by RWMutex.

func NewCatalog 

func NewCatalog() *Catalog

NewCatalog creates an empty catalog.

func (*Catalog) HasGraphWithPrefix 

func (c *Catalog) HasGraphWithPrefix(prefix string) bool

HasGraphWithPrefix returns true when any registered graph name starts with prefix. Names are stored lowercased; callers should pass a lowercased prefix.

func (*Catalog) ListGraphs 

func (c *Catalog) ListGraphs() []string

ListGraphs returns all registered graph names.

func (*Catalog) PopulateFromManager 

func (c *Catalog) PopulateFromManager(mgr *multidb.DatabaseManager) error

PopulateFromManager loads graph registrations from a DatabaseManager. It registers:

  • each standard database as a LocationLocal
  • each composite database as a LocationLocal
  • each constituent of a composite database as "composite.alias" with LocationLocal for local constituents or LocationRemote for remote ones

Previously registered graphs are cleared and replaced.

func (*Catalog) Register 

func (c *Catalog) Register(name string, loc Location)

Register adds or replaces a graph location in the catalog.

func (*Catalog) Resolve 

func (c *Catalog) Resolve(name string) (Location, error)

Resolve looks up a graph location by name. Returns an error if the graph is not registered.

func (*Catalog) Unregister 

func (c *Catalog) Unregister(name string)

Unregister removes a graph from the catalog.

type CommitCallback 

type CommitCallback func(sub *SubTransaction) error

CommitCallback is called for each sub-transaction during commit. The callback should perform the actual commit on the shard. If any callback returns an error, commit halts and the remaining sub-transactions are rolled back via RollbackCallback.

type CypherExecutor 

type CypherExecutor interface {
	// ExecuteQuery runs a Cypher query against a storage engine and returns columns + rows.
	ExecuteQuery(ctx context.Context, dbName string, engine storage.Engine, query string, params map[string]interface{}) ([]string, [][]interface{}, error)
	// ExecuteQueryWithRecord runs a Cypher query with correlated record bindings.
	ExecuteQueryWithRecord(ctx context.Context, dbName string, engine storage.Engine, query string, params map[string]interface{}, recordBindings map[string]interface{}) ([]string, [][]interface{}, error)
}

CypherExecutor is the interface for executing Cypher queries against a storage engine. It decouples the fabric package from the concrete cypher.StorageExecutor to avoid circular imports.

type FabricExecutor 

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

FabricExecutor walks a Fragment tree and dispatches each FragmentExec to either a local or remote executor based on the graph location resolved from the Catalog.

This mirrors Neo4j's FabricExecutor.java.

func NewFabricExecutor 

func NewFabricExecutor(catalog *Catalog, local *LocalFragmentExecutor, remote *RemoteFragmentExecutor) *FabricExecutor

NewFabricExecutor creates a fabric executor.

func (*FabricExecutor) Execute 

func (e *FabricExecutor) Execute(ctx context.Context, tx *FabricTransaction, fragment Fragment, params map[string]interface{}, authToken string) (*ResultStream, error)

Execute runs a Fragment tree within the context of a FabricTransaction.

Parameters:

  • ctx: context for cancellation/deadline propagation
  • tx: the distributed transaction (may be nil for auto-commit)
  • fragment: the root of the Fragment tree to execute
  • params: query parameters
  • authToken: the caller's auth token for OIDC forwarding to remote shards

type FabricPlanner 

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

FabricPlanner decomposes a Cypher query into a Fragment tree by splitting at USE-clause boundaries. Without USE clauses, it produces a single FragmentExec targeting the session database — identical to community behavior.

This mirrors Neo4j's FabricPlanner.scala.

func NewFabricPlanner 

func NewFabricPlanner(catalog *Catalog) *FabricPlanner

NewFabricPlanner creates a planner backed by the given catalog.

func (*FabricPlanner) Plan 

func (p *FabricPlanner) Plan(query string, sessionDB string) (Fragment, error)

Plan decomposes a query into a Fragment tree.

Parameters:

  • query: the full Cypher query string
  • sessionDB: the default database for the session (used when no USE clause is present)

Returns a Fragment tree ready for execution by FabricExecutor.

type FabricTransaction 

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

FabricTransaction coordinates sub-transactions across participating shards within a single distributed transaction.

Invariants enforced:

  • Reads may span any number of shards.
  • At most one shard may receive write operations per transaction.
  • Attempting a write on a second shard is rejected deterministically.
  • Commit commits all open sub-transactions; rollback rolls back all.

This mirrors Neo4j's FabricTransaction.java.

func FabricTransactionFromContext 

func FabricTransactionFromContext(ctx context.Context) (*FabricTransaction, bool)

FabricTransactionFromContext returns the active Fabric transaction from context.

func NewFabricTransaction 

func NewFabricTransaction(txID string) *FabricTransaction

NewFabricTransaction creates a new distributed transaction.

func (*FabricTransaction) BindParticipantCallbacks 

func (t *FabricTransaction) BindParticipantCallbacks(shardName string, commitFn CommitCallback, rollbackFn RollbackCallback) error

BindParticipantCallbacks binds per-subtransaction commit/rollback callbacks. Existing bindings are replaced.

func (*FabricTransaction) Commit 

func (t *FabricTransaction) Commit(commitFn CommitCallback, rollbackFn RollbackCallback) error

Commit commits all open sub-transactions using the provided callback. On partial failure, remaining sub-transactions are rolled back via compensation.

Limitation: this is not a full 2PC (two-phase commit) coordinator. Compensation rollback is best-effort — if a shard commit succeeds but a subsequent shard fails, the already-committed shard's changes cannot be truly undone (only compensated). Neo4j Fabric has the same fundamental constraint for cross-shard transactions. The many-read/one-write invariant mitigates this by limiting write exposure to a single shard per transaction.

func (*FabricTransaction) GetOrOpen 

func (t *FabricTransaction) GetOrOpen(shardName string, isWrite bool) (*SubTransaction, error)

GetOrOpen returns an existing sub-transaction for the shard, or opens a new one. If isWrite is true and a different shard already holds the write lock, ErrSecondWriteShard is returned.

func (*FabricTransaction) Participants 

func (t *FabricTransaction) Participants() []string

Participants returns the names of all shards participating in this transaction.

func (*FabricTransaction) Rollback 

func (t *FabricTransaction) Rollback(rollbackFn RollbackCallback) error

Rollback rolls back all open sub-transactions using the provided callback.

func (*FabricTransaction) State 

func (t *FabricTransaction) State() string

State returns the current transaction state.

func (*FabricTransaction) SubTransactions 

func (t *FabricTransaction) SubTransactions() map[string]*SubTransaction

SubTransactions returns a snapshot of all sub-transactions.

func (*FabricTransaction) TxID 

func (t *FabricTransaction) TxID() string

TxID returns the transaction identifier.

func (*FabricTransaction) WriteShard 

func (t *FabricTransaction) WriteShard() string

WriteShard returns the name of the shard that holds the write lock, or empty string if no writes have been performed.

type Fragment 

type Fragment interface {

	// OutputColumns returns the column names produced by this fragment.
	OutputColumns() []string
	// contains filtered or unexported methods
}

Fragment is a sealed interface representing a node in the query plan tree produced by decomposing a Cypher statement at USE-clause boundaries.

The Fragment tree is the core data structure of the fabric planner. It directly mirrors Neo4j's Fragment.scala ADT.

type FragmentApply 

type FragmentApply struct {
	// Input is the outer fragment producing driver rows.
	Input Fragment

	// Inner is the fragment executed once per input row.
	Inner Fragment

	// Columns lists the combined output columns from Input and Inner.
	Columns []string
}

FragmentApply represents sequential execution: for each row from Input, execute Inner and concatenate results. This models correlated subqueries where the inner query references variables from the outer scope.

func (*FragmentApply) OutputColumns 

func (f *FragmentApply) OutputColumns() []string

OutputColumns returns the apply's output columns.

type FragmentExec 

type FragmentExec struct {
	// Input is the fragment that feeds rows into this executable unit.
	Input Fragment

	// Query is the Cypher query string to execute against the target graph.
	Query string

	// GraphName identifies the target graph (e.g. "nornic.tr").
	// For dotted names, the part before the dot is the composite database
	// and the part after the dot is the constituent alias.
	GraphName string

	// Columns lists the output column names produced by this execution.
	Columns []string

	// IsWrite indicates whether this fragment performs write operations.
	IsWrite bool
}

FragmentExec is an executable fragment bound to a specific graph location. It represents a unit of work that can be dispatched to either a local or remote executor.

func (*FragmentExec) OutputColumns 

func (f *FragmentExec) OutputColumns() []string

OutputColumns returns the exec's output columns.

type FragmentInit 

type FragmentInit struct {
	// Columns lists the argument columns available to downstream fragments.
	Columns []string

	// ImportColumns lists columns imported from an outer scope (correlated subquery).
	ImportColumns []string
}

FragmentInit is the entry point of a Fragment tree. It produces a single empty row (like a SQL "dual" table) and defines the initial variable scope for downstream fragments.

func (*FragmentInit) OutputColumns 

func (f *FragmentInit) OutputColumns() []string

OutputColumns returns the initial columns.

type FragmentLeaf 

type FragmentLeaf struct {
	// Input is the fragment that feeds rows into this leaf.
	Input Fragment

	// Clauses contains the raw Cypher clause text.
	Clauses string

	// Columns lists the output column names produced by this leaf.
	Columns []string
}

FragmentLeaf is an intermediate fragment representing raw Cypher clauses before they are bound to a specific graph location.

func (*FragmentLeaf) OutputColumns 

func (f *FragmentLeaf) OutputColumns() []string

OutputColumns returns the leaf's output columns.

type FragmentUnion 

type FragmentUnion struct {
	// Init is the shared entry point for both branches.
	Init *FragmentInit

	// LHS is the left-hand branch.
	LHS Fragment

	// RHS is the right-hand branch.
	RHS Fragment

	// Distinct controls whether duplicate rows are eliminated.
	Distinct bool

	// Columns lists the output column names (must match between LHS and RHS).
	Columns []string
}

FragmentUnion represents parallel execution of two branches with result merging. If Distinct is true, duplicate rows are removed from the merged result.

func (*FragmentUnion) OutputColumns 

func (f *FragmentUnion) OutputColumns() []string

OutputColumns returns the union's output columns.

type HotPathTrace 

type HotPathTrace struct {
	ApplyBatchedLookupRows bool
}

HotPathTrace captures per-query Fabric optimization branch usage.

type LocalFragmentExecutor 

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

LocalFragmentExecutor executes fragments against a local storage engine via the existing Cypher executor infrastructure.

func NewLocalFragmentExecutor 

func NewLocalFragmentExecutor(cypherExec CypherExecutor, getEngine func(string) (storage.Engine, error)) *LocalFragmentExecutor

NewLocalFragmentExecutor creates a local executor.

Parameters:

  • cypherExec: the Cypher query executor
  • getEngine: function to resolve a database name to a storage.Engine

func (*LocalFragmentExecutor) Execute 

func (l *LocalFragmentExecutor) Execute(ctx context.Context, loc *LocationLocal, query string, params map[string]interface{}) (*ResultStream, error)

Execute runs a Cypher query against a local database.

func (*LocalFragmentExecutor) ExecuteRows 

func (l *LocalFragmentExecutor) ExecuteRows(ctx context.Context, loc *LocationLocal, query string, params map[string]interface{}) ([]string, RowIterator, error)

ExecuteRows runs a Cypher query and returns a row iterator.

func (*LocalFragmentExecutor) ExecuteWithRecord 

func (l *LocalFragmentExecutor) ExecuteWithRecord(ctx context.Context, loc *LocationLocal, query string, params map[string]interface{}, recordBindings map[string]interface{}) (*ResultStream, error)

ExecuteWithRecord runs a Cypher query against a local database with optional correlated bindings.

func (*LocalFragmentExecutor) ExecuteWithRecordRows 

func (l *LocalFragmentExecutor) ExecuteWithRecordRows(ctx context.Context, loc *LocationLocal, query string, params map[string]interface{}, recordBindings map[string]interface{}) ([]string, RowIterator, error)

ExecuteWithRecordRows runs a Cypher query with correlated bindings and returns a row iterator over the result.

type Location 

type Location interface {

	// DatabaseName returns the target database name.
	DatabaseName() string
	// contains filtered or unexported methods
}

Location represents where a FragmentExec runs. It mirrors Neo4j's Location.java sealed hierarchy.

type LocationLocal 

type LocationLocal struct {
	// DBName is the local database name.
	DBName string
}

LocationLocal indicates execution on the current NornicDB instance.

func (*LocationLocal) DatabaseName 

func (l *LocationLocal) DatabaseName() string

DatabaseName returns the local database name.

type LocationRemote 

type LocationRemote struct {
	// DBName is the database name on the remote instance.
	DBName string

	// URI is the remote endpoint URI (bolt://, neo4j://, http://, https://).
	URI string

	// AuthMode is the authentication mode: "oidc_forwarding" or "user_password".
	AuthMode string

	// User is the explicit username (only when AuthMode == "user_password").
	User string

	// Password is the explicit password (only when AuthMode == "user_password").
	// This is the decrypted plaintext resolved at runtime; never persisted.
	Password string
}

LocationRemote indicates execution on a remote NornicDB instance.

func (*LocationRemote) DatabaseName 

func (l *LocationRemote) DatabaseName() string

DatabaseName returns the remote database name.

type PlanCache 

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

PlanCache stores planned Fabric fragments keyed by normalized query + session DB. It uses LRU eviction and is safe for concurrent use.

func NewPlanCache 

func NewPlanCache(maxSize int) *PlanCache

NewPlanCache creates a new Fabric plan cache.

func (*PlanCache) Clear 

func (pc *PlanCache) Clear()

Clear removes all cached plans.

func (*PlanCache) Get 

func (pc *PlanCache) Get(query, sessionDB string) (Fragment, bool)

Get retrieves a cached fragment plan.

func (*PlanCache) Put 

func (pc *PlanCache) Put(query, sessionDB string, fragment Fragment)

Put stores a fragment plan in cache.

func (*PlanCache) Stats 

func (pc *PlanCache) Stats() (hits, misses int64, size int)

Stats returns cache hit/miss counters and current size.

type QueryGateway 

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

QueryGateway is the shared Fabric entrypoint used by protocol adapters. It owns planning + fragment execution so Bolt/HTTP/GraphQL can share one path.

func NewQueryGateway 

func NewQueryGateway(planner *FabricPlanner, executor *FabricExecutor) *QueryGateway

NewQueryGateway creates a gateway from planner+executor dependencies.

func (*QueryGateway) Execute 

func (g *QueryGateway) Execute(
	ctx context.Context,
	tx *FabricTransaction,
	query string,
	sessionDB string,
	params map[string]interface{},
	authToken string,
) (*ResultStream, error)

Execute plans and executes a Cypher query via Fabric semantics.

type RemoteFragmentExecutor 

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

RemoteFragmentExecutor executes fragments against a remote NornicDB instance via the RemoteEngine transport (Bolt or HTTP, auto-detected from URI scheme).

func NewRemoteFragmentExecutor 

func NewRemoteFragmentExecutor() *RemoteFragmentExecutor

NewRemoteFragmentExecutor creates a remote executor.

func (*RemoteFragmentExecutor) Close 

func (r *RemoteFragmentExecutor) Close() error

Close closes all cached remote engines.

func (*RemoteFragmentExecutor) Execute 

func (r *RemoteFragmentExecutor) Execute(ctx context.Context, loc *LocationRemote, query string, params map[string]interface{}, authToken string) (*ResultStream, error)

Execute runs a Cypher query against a remote NornicDB instance.

Parameters:

  • ctx: context for cancellation/deadline propagation
  • loc: the remote location (URI, database, auth config)
  • query: the Cypher query to execute
  • params: query parameters
  • authToken: the caller's auth token for OIDC forwarding

func (*RemoteFragmentExecutor) ExecuteRows 

func (r *RemoteFragmentExecutor) ExecuteRows(ctx context.Context, loc *LocationRemote, query string, params map[string]interface{}, authToken string) ([]string, RowIterator, error)

ExecuteRows runs a Cypher query against a remote location and returns a row iterator.

type ResultStream 

type ResultStream struct {
	// Columns is the ordered list of column names.
	Columns []string

	// Rows holds the result data. Each row has len(Columns) elements.
	Rows [][]interface{}
}

ResultStream holds the tabular output of a fragment execution. It carries column names and rows in the same shape as cypher.ExecuteResult.

func (*ResultStream) Empty 

func (r *ResultStream) Empty() bool

Empty returns true if the result stream has no rows.

func (*ResultStream) Merge 

func (r *ResultStream) Merge(other *ResultStream)

Merge appends the rows from other into this result stream. Columns must match. If this stream has no columns yet, they are adopted from other.

func (*ResultStream) RowCount 

func (r *ResultStream) RowCount() int

RowCount returns the number of rows.

type RollbackCallback 

type RollbackCallback func(sub *SubTransaction) error

RollbackCallback is called for each sub-transaction during rollback.

type RowIterator 

type RowIterator interface {
	Next() bool
	Row() []interface{}
	Err() error
	Close() error
}

RowIterator iterates result rows without requiring callers to index into a materialized [][]interface{} directly.

func NewConcatRowIterator 

func NewConcatRowIterator(iterators ...RowIterator) RowIterator

NewConcatRowIterator returns a row iterator that reads iterators in order.

func NewConvertingRowIterator 

func NewConvertingRowIterator(base RowIterator, convert func([]interface{}) []interface{}) RowIterator

NewConvertingRowIterator wraps an iterator and lazily converts each row on Row().

func NewDistinctRowIterator 

func NewDistinctRowIterator(base RowIterator) RowIterator

NewDistinctRowIterator filters duplicate rows while streaming.

func NewPrefetchRowIterator 

func NewPrefetchRowIterator(ctx context.Context, base RowIterator, buffer int) RowIterator

NewPrefetchRowIterator creates a bounded prefetch iterator with backpressure.

func NewResultRowIterator 

func NewResultRowIterator(r *ResultStream) RowIterator

NewResultRowIterator wraps a ResultStream as a RowIterator.

type RowView 

type RowView interface {
	Len() int
	At(i int) interface{}
	Materialize() []interface{}
}

RowView exposes row values lazily by index.

func NewJoinedRowView 

func NewJoinedRowView(outer []interface{}, inner []interface{}, fromOuter []int, fromInner []int) RowView

NewJoinedRowView creates a lazy joined row view for precomputed column mappings.

func NewSliceRowView 

func NewSliceRowView(row []interface{}) RowView

type SubTransaction 

type SubTransaction struct {
	// ShardName identifies the constituent (e.g. "nornic.tr").
	ShardName string

	// IsWrite is true if any write has been performed on this shard.
	IsWrite bool

	// State tracks the lifecycle: "open", "committed", "rolledback".
	State string
	// contains filtered or unexported fields
}

SubTransaction represents an open sub-transaction on a single shard.

func SubTransactionFromContext 

func SubTransactionFromContext(ctx context.Context) (*SubTransaction, bool)

SubTransactionFromContext returns the active fabric sub-transaction from context.

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.