Source File
	stream.go

Belonging Package
	github.com/freeeve/gochickpeas/gql/internal/exec

// The push pipeline: a segment's stages run as a chain of row sinks, each
// holding one reused full-width row buffer, so matched rows flow through
// the chain without per-row allocation. Only the terminal sink retains
// rows (in chunked arenas); the shortest-path stage keeps its batch
// semantics behind a buffering barrier. Depth-first push order equals the
// former materialize-then-iterate order, so group encounter order,
// DISTINCT first-occurrence, and stable-sort ties are unchanged.
package exec

import (
	chickpeas "github.com/freeeve/gochickpeas"
	"github.com/freeeve/gochickpeas/gql/internal/eval"
	"github.com/freeeve/gochickpeas/gql/internal/graph"
	"github.com/freeeve/gochickpeas/gql/internal/plan"
	"github.com/freeeve/gochickpeas/gql/value"
)

// rowSink consumes full-width rows pushed through a segment's stage chain.
// A pushed row is valid only for the duration of the call; a sink that
// retains it must copy. push reports whether the consumer wants more rows:
// false is the stop protocol -- an unordered LIMIT that has its k rows
// tells every producer above it to abandon the walk (aggregation and
// ORDER BY must consume everything and always report true). A producer
// receiving false stops emitting and propagates false after its own
// cleanup. close flushes any buffered state downstream.
type rowSink interface {
	push(row []value.Value) bool
	close()
}

// arenaChunkValues caps the bump-arena chunk size in values (not rows);
// arenaChunkStart seeds the geometric ladder.
const (
	arenaChunkValues = 16384
	arenaChunkStart  = 512
)

// rowArena bump-allocates fixed-width rows out of large chunks, so
// retaining n rows costs n/chunk allocations instead of n. A sink that
// knows its retention is bounded sets chunkValues so a small bound does
// not pay a full-size chunk.
type rowArena struct {
	width       int
	chunkValues int // fixed chunk size in values; 0 means geometric growth
	// nextChunk is the geometric ladder state: chunks start small and
	// double up to arenaChunkValues, so a sink retaining a handful of
	// rows pays kilobytes rather than a near-empty full-size chunk,
	// while a large retention converges on the same big chunks after a
	// few (amortized-free) steps.
	nextChunk int
	chunk     []value.Value
	off       int
}

// alloc returns the next zeroed width-wide row.
func (a *rowArena) alloc() []value.Value {
	if a.off+a.width > len(a.chunk) {
		cv := a.chunkValues
		if cv == 0 {
			if a.nextChunk == 0 {
				a.nextChunk = arenaChunkStart
			}
			cv = a.nextChunk
			if a.nextChunk < arenaChunkValues {
				a.nextChunk *= 2
			}
		}
		n := max(cv, a.width)
		a.chunk = make([]value.Value, n)
		a.off = 0
	}
	r := a.chunk[a.off : a.off+a.width : a.off+a.width]
	a.off += a.width
	return r
}

// rollback releases the arena's most recent alloc (a DISTINCT duplicate).
func (a *rowArena) rollback() {
	a.off -= a.width
}

// copyRow retains a transient row in the arena.
func (a *rowArena) copyRow(row []value.Value) []value.Value {
	r := a.alloc()
	copy(r, row)
	return r
}

// matchSink runs one MATCH stage per pushed row: genMatches walks the bind
// chain over the stage-local buffer, forwarding each bound match. A named
// path is assembled and post-path-filtered per match; an OPTIONAL stage
// that produced nothing re-emits the input row from the orig buffer.
type matchSink struct {
	ctx         *eval.Ctx
	stage       *plan.MatchStage
	comp        *stageComp
	slots       map[string]int
	buf, orig   []value.Value
	scratch     genScratch
	next        rowSink
	opRows      []uint64
	profScratch []uint64
	fired       bool
	pathFilters []RowEval
	// uniq is the segment chain's shared used-relationship env: chained
	// MATCH stages from one clause (comma patterns, planner splits) see
	// one stack per in-flight row.
	uniq *uniqEnv
	// emitFn is the emit method bound once, so genMatches gets the same
	// closure every push instead of a fresh method value.
	emitFn func([]value.Value) bool
}

func (m *matchSink) push(row []value.Value) bool {
	// Absorbed-chain gates on carried slots: a failing row cannot match
	// the stage (the absorber refuses OPTIONAL stages, so dropping is
	// exact).
	for i := range m.comp.gates {
		if !m.comp.gates[i].pass(row) {
			return true
		}
	}
	copy(m.buf, row)
	if m.stage.Optional {
		copy(m.orig, row)
		m.fired = false
		more := genMatches(m.ctx, m.comp.ops, m.buf, m.comp, m.slots, m.uniq, m.emitFn, &m.scratch, m.profRows())
		m.scatterProf()
		if more && !m.fired {
			// The re-emitted row takes the path assembly and post-path
			// WHERE too, exactly like the former batch bindPaths pass.
			return m.forward(m.orig)
		}
		return more
	}
	more := genMatches(m.ctx, m.comp.ops, m.buf, m.comp, m.slots, m.uniq, m.emitFn, &m.scratch, m.profRows())
	m.scatterProf()
	return more
}

// profRows is the opRows slice genMatches writes: the stage's own when
// exec ops match the plan's, else a scratch sized to the exec ops that
// scatterProf redistributes -- PROFILE lines index ORIGINAL operators,
// and an absorbed chain's ops must report 0, not another op's count.
func (m *matchSink) profRows() []uint64 {
	if m.opRows == nil || m.comp.profMap == nil {
		return m.opRows
	}
	if m.profScratch == nil {
		m.profScratch = make([]uint64, len(m.comp.ops)+1)
	}
	clear(m.profScratch)
	return m.profScratch
}

// scatterProf folds a scratch profile back onto the original op lines.
func (m *matchSink) scatterProf() {
	if m.opRows == nil || m.comp.profMap == nil {
		return
	}
	for e, orig := range m.comp.profMap {
		m.opRows[orig] += m.profScratch[e]
	}
	// The trailing stage-WHERE tally.
	m.opRows[len(m.opRows)-1] += m.profScratch[len(m.profScratch)-1]
}

// emit forwards one bound match. The OPTIONAL no-match probe counts every
// match (like the former pre-bindPaths batch check), so it flips before
// the post-path filter can prune.
func (m *matchSink) emit(r []value.Value) bool {
	m.fired = true
	return m.forward(r)
}

// forward assembles the row's named path and applies the post-path WHERE
// conjuncts, then hands the row downstream.
func (m *matchSink) forward(r []value.Value) bool {
	if pb := m.stage.PathBind; pb != nil {
		rels := pathRelPositionsOf(r[pb.RelsSlot])
		var nodes []graph.NodeID
		if start, ok := r[pb.FromSlot].AsNode(); ok {
			nodes = reconstructPathNodes(m.ctx, start, rels)
		}
		r[pb.PathSlot] = value.Path(nodes, rels)
		for _, f := range m.pathFilters {
			if !f.Eval(m.ctx, r, m.slots).IsTruthy() {
				return true
			}
		}
	}
	return m.next.push(r)
}

func (m *matchSink) close() { m.next.close() }

// unwindSink is FOR x IN list per pushed row: a list emits one row per
// element, null emits none, any other scalar emits a single row.
type unwindSink struct {
	ctx   *eval.Ctx
	list  RowEval
	slots map[string]int
	out   int
	buf   []value.Value
	next  rowSink
	count *uint64
}

func (u *unwindSink) push(row []value.Value) bool {
	v := u.list.Eval(u.ctx, row, u.slots)
	if items, ok := v.AsList(); ok {
		for _, item := range items {
			copy(u.buf, row)
			u.buf[u.out] = item
			u.bump()
			if !u.next.push(u.buf) {
				return false
			}
		}
		return true
	}
	if v.IsNull() {
		return true
	}
	copy(u.buf, row)
	u.buf[u.out] = v
	u.bump()
	return u.next.push(u.buf)
}

func (u *unwindSink) bump() {
	if u.count != nil {
		*u.count++
	}
}

func (u *unwindSink) close() { u.next.close() }

// callSink crosses each pushed row with a procedure's results -- computed
// once at build for a constant-argument call (per-node scalar vector,
// index search hit-set, or propagation rows), or per pushed row for a
// correlated call (argEvals non-nil), where a row whose evaluated
// arguments fail validation yields no rows. A backend without the native
// capability passes rows through unchanged.
type callSink struct {
	ctx    *eval.Ctx
	cs     *plan.CallStage
	values []value.Value
	// hits is the search hit-set's ascending-id iteration (nil = no hits);
	// unnamed so the nodeset iterator assigns directly.
	hits func(yield func(graph.NodeID) bool)
	// prop is the resolved algo.propagate result set.
	prop     []chickpeas.PropagateResult
	argEvals []RowEval
	slots    map[string]int
	g        *chickpeas.Snapshot
	native   bool
	buf      []value.Value
	next     rowSink
	count    *uint64
}

func (c *callSink) push(row []value.Value) bool {
	if !c.native {
		c.bump()
		return c.next.push(row)
	}
	values, hits, prop := c.values, c.hits, c.prop
	if c.argEvals != nil {
		args := make([]value.Value, len(c.argEvals))
		for i, ev := range c.argEvals {
			args[i] = ev.Eval(c.ctx, row, c.slots)
		}
		proc, err := plan.ResolveCallProc(c.cs.ProcName, args)
		if err != nil {
			return true
		}
		values, hits, prop = callResults(&proc, c.g)
	}
	if c.cs.Proc.Kind == plan.ProcPropagate {
		for _, r := range prop {
			copy(c.buf, row)
			if c.cs.NodeSlot != plan.NoSlot {
				c.buf[c.cs.NodeSlot] = value.Node(r.Node)
			}
			if c.cs.ValueSlot != plan.NoSlot {
				c.buf[c.cs.ValueSlot] = value.Float(r.Value)
			}
			if c.cs.DepthSlot != plan.NoSlot {
				c.buf[c.cs.DepthSlot] = value.Int(int64(r.Depth))
			}
			c.bump()
			if !c.next.push(c.buf) {
				return false
			}
		}
		return true
	}
	if values != nil {
		// The per-node vector is id-space-sized, so on a sparse space its
		// indexes include gap ids; the existence oracle keeps phantom
		// rows out of the result (a real node structurally cannot drop --
		// the vector covers the whole space).
		for i, v := range values {
			if !c.g.NodeExists(chickpeas.NodeID(i)) {
				continue
			}
			copy(c.buf, row)
			if c.cs.NodeSlot != plan.NoSlot {
				c.buf[c.cs.NodeSlot] = value.Node(graph.NodeID(i))
			}
			if c.cs.ValueSlot != plan.NoSlot {
				c.buf[c.cs.ValueSlot] = v
			}
			c.bump()
			if !c.next.push(c.buf) {
				return false
			}
		}
		return true
	}
	if hits == nil {
		return true
	}
	more := true
	hits(func(nid graph.NodeID) bool {
		copy(c.buf, row)
		if c.cs.NodeSlot != plan.NoSlot {
			c.buf[c.cs.NodeSlot] = value.Node(nid)
		}
		c.bump()
		more = c.next.push(c.buf)
		return more
	})
	return more
}

func (c *callSink) bump() {
	if c.count != nil {
		*c.count++
	}
}

func (c *callSink) close() { c.next.close() }

// subquerySink is CALL { } per pushed row: a correlated subquery runs the
// sub-plan seeded from the outer row's imports; an uncorrelated one runs
// once on first use and cross-joins.
type subquerySink struct {
	ctx     *eval.Ctx
	cs      *plan.CallSubqueryStage
	buf     []value.Value
	seed    []value.Value
	subRows [][]value.Value
	subRun  bool
	next    rowSink
	count   *uint64
}

func (s *subquerySink) push(row []value.Value) bool {
	var sub [][]value.Value
	if len(s.cs.ImportSlots) == 0 {
		if !s.subRun {
			s.subRows = runSubplan(s.ctx, s.cs.Sub, nil)
			s.subRun = true
		}
		sub = s.subRows
	} else {
		for i, slot := range s.cs.ImportSlots {
			s.seed[i] = row[slot]
		}
		sub = runSubplan(s.ctx, s.cs.Sub, s.seed)
	}
	for _, sr := range sub {
		copy(s.buf, row)
		for i, slot := range s.cs.OutSlots {
			s.buf[slot] = sr[i]
		}
		if s.count != nil {
			*s.count++
		}
		if !s.next.push(s.buf) {
			return false
		}
	}
	return true
}

func (s *subquerySink) close() { s.next.close() }

// spSink is the shortest-path stage's batch barrier: input rows buffer in
// an arena until close, because the batch runner memoizes a parent tree
// for sources shared by multiple rows.
type spSink struct {
	ctx   *eval.Ctx
	sp    *plan.SpStage
	rows  [][]value.Value
	arena rowArena
	next  rowSink
	count *uint64
}

func (s *spSink) push(row []value.Value) bool {
	s.rows = append(s.rows, s.arena.copyRow(row))
	return true
}

func (s *spSink) close() {
	out := runSPStage(s.ctx, s.sp, s.rows)
	if s.count != nil {
		*s.count = uint64(len(out))
	}
	for _, r := range out {
		if !s.next.push(r) {
			break
		}
	}
	s.next.close()
}

// gateSink is an injected early row gate (plan.GateStage): like spSink it
// buffers its input so the batch path runner keeps its shared-source tree
// memo, then binds the path into its hidden slot, evaluates the hoisted
// LET chain, and forwards only rows passing the hoisted filter. Rows it
// drops are exactly rows the downstream chain would drop (the search is
// deterministic per endpoint pair), so results are identical -- just
// cheaper, because survivors alone reach the stages behind it.
type gateSink struct {
	ctx     *eval.Ctx
	gs      *plan.GateStage
	slots   map[string]int
	derived []RowEval
	where   RowEval
	rows    [][]value.Value
	arena   rowArena
	next    rowSink
	count   *uint64
}

func (s *gateSink) push(row []value.Value) bool {
	s.rows = append(s.rows, s.arena.copyRow(row))
	return true
}

func (s *gateSink) close() {
	out := runSPStage(s.ctx, &s.gs.Sp, s.rows)
	for _, r := range out {
		for i, d := range s.gs.Derived {
			r[d.Slot] = s.derived[i].Eval(s.ctx, r, s.slots)
		}
		if !s.where.Eval(s.ctx, r, s.slots).IsTruthy() {
			continue
		}
		if s.count != nil {
			*s.count++
		}
		if !s.next.push(r) {
			break
		}
	}
	s.next.close()
}

The pages are generated with Golds v0.8.7. (GOOS=linux GOARCH=amd64)
Golds is a Go 101 project developed by Tapir Liu.
PR and bug reports are welcome and can be submitted to the issue list.
Please follow @zigo_101 (reachable from the left QR code) to get the latest news of Golds.