// The bind-chain walker: iterative DFS over a MATCH stage's ops, binding
// each level's candidates into the row and pruning with the pushed-down
// conjuncts (port of the Rust gen_matches). M15 executes scan ops;
// expansion ops land in M17 and are rejected at plan validation.
package exec

import (
	chickpeas 
	
	
	
	
)

// genScratch holds the per-row DFS buffers, reused across a stage's row
// loop so a scan/expand does not allocate per row. walk and dedup serve the
// bounded trail enumeration the same way reach serves the reachability BFS
// (one live walk per stage at a time).
type genScratch struct {
	cand      [][]graph.NodeID
	candRel   [][]uint32
	candData  [][]uint32
	candRange [][][2]int
	// candPairData/candPairRange carry a CONTRIBUTING var-expand's
	// per-trail uniqueness pairs (flat arena + (start, len) per
	// candidate), pushed onto the row's used stack while the candidate is
	// bound.
	candPairData  [][][2]graph.NodeID
	candPairRange [][][2]int
	// uniqPushed counts each level's live pair pushes, retired on the
	// level's next bind, exhaustion, or row end.
	uniqPushed []int
	pos        []int
	reach      reachScratch
	walk       varWalk
	dedup      map[graph.NodeID]struct{}
	semiBuf    []graph.NodeID
	// swept marks levels whose specialized predicates already ran over
	// the freshly filled candidate buffer (the fill-time sweep), so the
	// pop loop skips them and counts bindings from the sweep's credit.
	swept []bool
	// scanMemo/scanMemoOK cache a level's row-independent fresh-scan
	// candidates across match calls: a constant-anchored scan nested
	// below the driving level otherwise recomputes an identical set for
	// every outer row. The memo holds the pristine post-matcher list
	// (filled before the per-row sweep/prunes mutate the working buffer)
	// and lives as long as this stage's stream node, matching the
	// fixed-per-execution parameters the scan can read.
	scanMemo   [][]graph.NodeID
	scanMemoOK []bool
	// keep is the sweep's columnar mask, reused across levels.
	keep []bool
	// chainRoots / chainFunc cache each var-expand op's chain-collapse
	// and functionality resolutions for this execution (presence =
	// checked). Plan ops are shared across cached executions, so the
	// resolutions cannot live on the op itself.
	chainRoots map[*plan.BindOp]chickpeas.RootsVia
	chainFunc  map[*plan.BindOp]bool
	// seedFrontier/seedNext are the EXISTS-seed walk's level buffers.
	seedFrontier []graph.NodeID
	seedNext     []graph.NodeID
}

// chainRootsFor resolves (once per op per execution) whether op's
// reachable walk collapses to a root-array lookup.
func ( *genScratch) ( *eval.Ctx,  *plan.BindOp) (chickpeas.RootsVia, bool) {
	if .chainRoots == nil {
		.chainRoots = map[*plan.BindOp]chickpeas.RootsVia{}
	}
	,  := .chainRoots[]
	if ! {
		, _ = .G.ChainRootsVia(.Types, .Dir, .Labels)
		.chainRoots[] = 
	}
	return ,  != nil
}

// chainFuncFor resolves (once per op per execution) whether op's rel
// type is functional in its direction.
func ( *genScratch) ( *eval.Ctx,  *plan.BindOp) bool {
	if .chainFunc == nil {
		.chainFunc = map[*plan.BindOp]bool{}
	}
	,  := .chainFunc[]
	if ! {
		 = .Dir != chickpeas.Both && .G.FunctionalVia(.Types, .Dir)
		.chainFunc[] = 
	}
	return 
}

// reachScratch holds varReach's dedup'd-BFS working sets, reused across a
// stage's row loop (the walk runs to completion per call and is never
// nested, so a single set serves every var-length reach in the stage).
type reachScratch struct {
	expanded map[graph.NodeID]struct{}
	emitted  map[graph.NodeID]struct{}
	frontier []graph.NodeID
	next     []graph.NodeID
	nbuf     []graph.NodeID
	// pbuf carries rel positions parallel to nbuf when the hop gate's
	// predicate needs them.
	pbuf []uint32
}

// genMatches walks the ops' bind chain over one input row, handing each
// completed match row to the sink by reference (the sink copies). opRows
// is PROFILE's counter slice (one slot per op for bindings produced, plus
// a final slot for rows passing the stage WHERE); nil when not profiling.
// Reports the sink's keep-going verdict: on a stop the walk abandons its
// remaining candidates, retiring every level's pair pushes on the way out
// so the shared uniqueness env is exactly as empty as a completed walk
// leaves it.
func genMatches( *eval.Ctx,  []plan.BindOp,  []value.Value,  *stageComp,  map[string]int,  *uniqEnv,  func([]value.Value) bool,  *genScratch,  []uint64) bool {
	// New match-call epoch: a loop-invariant carried IN list hashes once
	// for this call and reuses it across the call's candidates.
	.MatchEpoch++
	 := len()
	if  == 0 {
		 := ()
		if  != nil {
			[0]++
		}
		return 
	}
	for len(.cand) <  {
		.cand = append(.cand, nil)
		.candRel = append(.candRel, nil)
		.candData = append(.candData, nil)
		.candRange = append(.candRange, nil)
		.candPairData = append(.candPairData, nil)
		.candPairRange = append(.candPairRange, nil)
		.scanMemo = append(.scanMemo, nil)
		.scanMemoOK = append(.scanMemoOK, false)
	}
	.pos = append(.pos[:0], make([]int, )...)
	.uniqPushed = append(.uniqPushed[:0], make([]int, )...)
	.swept = append(.swept[:0], make([]bool, )...)
	// retire pops a level's live pair pushes off the used stack.
	 := func( int) {
		if .uniqPushed[] > 0 {
			.stack = .stack[:len(.stack)-.uniqPushed[]]
			.uniqPushed[] = 0
		}
	}
	 := 

	levelCandidates(, &[0], , 0, , , )
	pruneSemijoinLookahead(, , , 0, , )
	.swept[0] = sweepLevel(, &[0], , 0, , , )
	 := 0
	for {
		switch {
		case .pos[] < len(.cand[]):
			 := .pos[]
			 := .cand[][]
			.pos[]++
			// MATCH-scope rel uniqueness: retire the PREVIOUS candidate's
			// pair pushes at this level, then check/push this candidate's.
			// An Expand keys the single hop (from, node); a tracked
			// var-length op's per-trail pairs were collected at gather time
			// (with the scope's used pairs already excluded during the
			// walk, so only the push remains).
			()
			if  := [].Uniq;  != nil {
				switch [].Kind {
				case plan.OpExpand:
					if ,  := [[].From].AsNode();  {
						,  := uniqPair([].Dir, , )
						if .Check && .used(.Scope, , ) {
							continue
						}
						// Capture mode (hash-join build) also records a
						// Check-only op's pair as a dead entry, for the
						// probe's replay against the outer env.
						if .Contribute || (.capture && .Check) {
							.stack = append(.stack, uniqKey{scope: .Scope, a: , b: , dead: !.Contribute, check: .Check})
							.uniqPushed[] = 1
						}
					}
				case plan.OpVarExpand:
					if .Contribute {
						 := .candPairRange[][]
						for ,  := range .candPairData[][[0] : [0]+[1]] {
							.stack = append(.stack, uniqKey{scope: .Scope, a: [0], b: [1], check: .Check})
						}
						.uniqPushed[] = [1]
					}
				}
			}
			[slotOf(&[])] = value.Node()
			// A named fixed-hop relationship binds its position; a named
			// variable-length relationship binds the trail's rel list.
			if [].Kind == plan.OpExpand && [].RelSlot != plan.NoSlot {
				[[].RelSlot] = value.Rel(.candRel[][])
			}
			if [].Kind == plan.OpVarExpand && [].RelSlot != plan.NoSlot {
				 := .candRange[][]
				 := make([]value.Value, [1])
				for  := range [1] {
					[] = value.Rel(.candData[][[0]+])
				}
				[[].RelSlot] = value.List()
			}
			// PROFILE: the binding counts before the level filters prune,
			// so pushdown effectiveness is visible per op (a swept level
			// credited its pre-sweep volume at fill time).
			if  != nil && !.swept[] {
				[]++
			}
			// Pushed-down predicates for this level: any failing conjunct
			// abandons the candidate before deeper ops expand from it.
			// Specialized per-candidate predicates ran at fill time on a
			// swept level and run here otherwise; then the general filters.
			 := true
			if !.swept[] {
				for ,  := range .levelPreds[] {
					if !(, , ) {
						 = false
						break
					}
				}
			}
			if  {
				for ,  := range .levelFilters[] {
					if !.Eval(, , ).IsTruthy() {
						 = false
						break
					}
				}
			}
			if ! {
				continue
			}
			if +1 ==  {
				 := ()
				if  != nil {
					[]++
				}
				if ! {
					// Stop: unwind every live level's pair pushes exactly
					// as the exhausted path below would, then tell the
					// caller to stop too.
					for ;  > 0; -- {
						()
					}
					(0)
					return false
				}
			} else {
				++
				levelCandidates(, &[], , , , , )
				pruneSemijoinLookahead(, , , , , )
				.swept[] = sweepLevel(, &[], , , , , )
				.pos[] = 0
			}
		case  == 0:
			// Retire the last candidate's pair pushes so the env is empty
			// for the next row.
			(0)
			return true
		default:
			()
			--
		}
	}
}

// sweepLevel runs a level's specialized predicates over the freshly
// filled candidate buffer, compacting survivors in place (parallel
// rel-position and range buffers stay in sync) and crediting the level's
// PROFILE binding count with the pre-sweep volume. Every buffered
// candidate is popped exactly once and the DFS never exits early, so
// fill-time counting equals the former pop-time counting EXCEPT when the
// op is uniqueness-tracked -- a uniq-rejected candidate must keep not
// counting -- so tracked ops keep the pop-time path (reported false).
func sweepLevel( *eval.Ctx,  *plan.BindOp,  *stageComp,  int,  []value.Value,  *genScratch,  []uint64) bool {
	 := .levelPreds[]
	 := .levelBatch[]
	if (len() == 0 && len() == 0) || .Uniq != nil {
		return false
	}
	 := .cand[]
	if  != nil {
		[] += uint64(len())
	}
	// Columnar conjuncts first: each sweeps the whole buffer as one
	// typed pass over a keep mask, then a single compaction applies the
	// remaining per-candidate predicates to the survivors.
	var  []bool
	if len() > 0 {
		if cap(.keep) < len() {
			.keep = make([]bool, len())
		}
		 = .keep[:len()]
		for  := range  {
			[] = true
		}
		for ,  := range  {
			(, , , )
		}
	}
	 := .candRel[]
	 := .candRange[]
	 := len() == len()
	 := len() == len()
	 := 0
	for ,  := range  {
		if  != nil && ![] {
			continue
		}
		 := true
		for ,  := range  {
			if !(, , ) {
				 = false
				break
			}
		}
		if ! {
			continue
		}
		[] = 
		if  {
			[] = []
		}
		if  {
			[] = []
		}
		++
	}
	.cand[] = [:]
	if  {
		.candRel[] = [:]
	}
	if  {
		.candRange[] = [:]
	}
	return true
}

// slotOf is the node slot an op binds.
func slotOf( *plan.BindOp) int {
	if .Kind == plan.OpScan {
		return .Slot
	}
	return .To
}

// relSlotOf is the slot an op's relationship binding occupies -- a named
// fixed-length expand's single relationship or a named variable-length
// expand's relationship list; NoSlot otherwise. Pushdown placement keys
// on this: a conjunct reading a relationship slot missing from the level
// map would land where the slot is still null and silently drop rows.
func relSlotOf( *plan.BindOp) int {
	if .Kind == plan.OpExpand || .Kind == plan.OpVarExpand {
		return .RelSlot
	}
	return plan.NoSlot
}

// levelCandidates fills the pooled candidate buffers for binding an op's
// node slot (and, for named relationships, its rel position buffers).
func levelCandidates( *eval.Ctx,  *plan.BindOp,  *stageComp,  int,  []value.Value,  *uniqEnv,  *genScratch) {
	 := .matchers[]
	 := &.cand[]
	* = (*)[:0]
	.candRel[] = .candRel[][:0]
	.candData[] = .candData[][:0]
	.candRange[] = .candRange[][:0]
	.candPairData[] = .candPairData[][:0]
	.candPairRange[] = .candPairRange[][:0]
	switch .Kind {
	case plan.OpExpand:
		if .semijoins[] != nil {
			semijoinCandidates(, , , .relMatchers[], .semijoins[], , , &.semiBuf)
			return
		}
		expandCandidates(, , , .relMatchers[], , , &.candRel[])
		return
	case plan.OpVarExpand:
		varExpandCandidates(, , , .relMatchers[], .hopGates[], , , , &.candData[], &.candRange[], &.candPairData[], &.candPairRange[], )
		return
	}
	switch .Source.Kind {
	case plan.ScanArg:
		// The node is already bound in a row slot (carried in / reused
		// across MATCH clauses); re-test this op's own constraints.
		if ,  := [.Source.Slot].AsNode();  && .G.NodeMatcherAccepts(, ) {
			* = append(*, )
		}
	case plan.ScanNodeIDVar:
		// Per-row id seek: the slot holds the target id.
		if ,  := nodeIDSeekValue(, [.Source.Slot]);  && .G.NodeMatcherAccepts(, ) {
			* = append(*, )
		}
	case plan.ScanExistsSeed:
		// EXISTS-driven candidate superset; past the fan-out cap fall
		// back to the base source (the kept conjunct finalizes either
		// way).
		if !existsSeedCandidates(, , , .seedRel[], .seedNode[], , , ) {
			 := .Source
			.Kind = baseScanKind(&.Source)
			freshScan(, &, , false, )
		}
	default:
		// Row-independent sources (property/label/IN/id-literal/text/all
		// scans) yield the same candidates for every row of this stage,
		// so the first fill is memoized and later calls copy it instead
		// of re-running the scan -- the per-row sweep and prunes mutate
		// only the working buffer, never the memo.
		if disableScanMemo {
			freshScan(, &.Source, , scanMatcherRedundant(), )
			return
		}
		if .scanMemoOK[] {
			* = append(*, .scanMemo[]...)
			scanMemoHits++
			return
		}
		freshScan(, &.Source, , scanMatcherRedundant(), )
		.scanMemo[] = append(.scanMemo[][:0], *...)
		.scanMemoOK[] = true
	}
}

// scanMemoHits counts row-independent scan fills served from the memo,
// so tests can assert the reuse actually engaged.
var scanMemoHits int

// disableScanMemo pins result identity in tests: every match call must
// produce exactly the rows the memoized path does.
var disableScanMemo bool

// baseScanKind is the source a ScanExistsSeed degrades to when its walk
// is abandoned: the label scan when one exists, else every node.
func baseScanKind( *plan.ScanSource) plan.ScanKind {
	if .Label != "" {
		return plan.ScanLabel
	}
	return plan.ScanAll
}