// Hash-join stage runner: build the independent branch once per distinct
// external-slot tuple (an isolated capture-mode uniqueness env records
// each row's relationship pairs), then per outer row expand the probe hop
// and emit the key-matching payload rows -- replaying each payload's
// captured Check pairs against the outer used-pair env and pushing its
// live pairs for downstream ops, so the emitted multiset is exactly what
// the original nested execution produced.
package exec

import (
	

	
	
	
	
	
	
)

// hjPair is one captured relationship-uniqueness pair of a payload row:
// check pairs replay against the outer env at probe time, live pairs push
// onto it while downstream ops run.
type hjPair struct {
	scope       uint32
	a, b        graph.NodeID
	check, live bool
}

// hjRow is one materialized build row: its payload-slot values and its
// captured pairs (capped sub-slices of the table's append-only slabs),
// plus the intrusive link to the next row sharing its join key.
type hjRow struct {
	vals  []value.Value
	pairs []hjPair
	next  int32
}

// hjTable is one build's result, indexed by the key slot's node (expand-
// keyed joins) or the key expression's encoded value (value-keyed joins).
// Rows chain per key through hjRow.next in insertion order (heads/tails
// hold each chain's ends), and payload values and pairs pack into shared
// slabs -- the per-row map-bucket appends and slice makes the map-of-
// slices form paid were the build sink's whole allocation profile.
type hjTable struct {
	rows     []hjRow
	valSlab  []value.Value
	pairSlab []hjPair
	byKey    flatset.U64Map
	byVal    flatset.ByteMap
	heads    []int32
	tails    []int32
}

// mintChain opens an empty chain slot (the GetOrCreate create hook).
func ( *hjTable) () int {
	.heads = append(.heads, -1)
	.tails = append(.tails, -1)
	return len(.heads) - 1
}

// link appends row idx to chain ci, preserving insertion order.
func ( *hjTable) ( int,  int32) {
	if .tails[] < 0 {
		.heads[] = 
	} else {
		.rows[.tails[]].next = 
	}
	.tails[] = 
}

// headKey is chain head for an expand key; -1 when the key has no rows.
func ( *hjTable) ( graph.NodeID) int32 {
	if ,  := .byKey.Get(uint64());  {
		return .heads[]
	}
	return -1
}

// headVal is chain head for an encoded value key; -1 when absent.
func ( *hjTable) ( []byte) int32 {
	if ,  := .byVal.Get();  {
		return .heads[]
	}
	return -1
}

// hashJoinSink is the stage's row sink.
type hashJoinSink struct {
	ctx     *eval.Ctx
	hj      *plan.HashJoinStage
	seg     *plan.Segment
	slots   map[string]int
	buf     []value.Value
	next    rowSink
	uniq    *uniqEnv
	probeM  *graph.NodeMatcher
	probeRM *graph.RelMatcher
	where   []RowEval
	tables  map[string]*hjTable
	keyBuf  []byte
	nbuf    []graph.NodeID
	count   *uint64
	// keyProbe is the value-keyed join's outer-side key (nil for the
	// expand-keyed form); valBuf is its encoding scratch.
	keyProbe RowEval
	valBuf   []byte
}

func newHashJoinSink( *eval.Ctx,  *plan.Segment,  *plan.HashJoinStage,  rowSink,  *uniqEnv,  *uint64) *hashJoinSink {
	 := make([]graph.PropSpec, len(.Probe.Props))
	for ,  := range .Probe.Props {
		[] = graph.PropSpec{Key: .Key, Val: eval.LitValue(, .Val)}
	}
	 := &hashJoinSink{
		ctx: , hj: , seg: , slots: .Slots,
		buf: make([]value.Value, .RowWidth), next: , uniq: ,
		probeM:  .G.CompileNodeMatcher(.Probe.Labels, ),
		probeRM: .G.CompileRelMatcher(.Probe.Types),
		tables:  map[string]*hjTable{},
		count:   ,
	}
	if .Where != nil {
		var  []ast.Expr
		plan.SplitAnd(.Where, &)
		for ,  := range  {
			.where = append(.where, compileEval(, , .Slots))
		}
	}
	if .KeyProbe != nil {
		.keyProbe = compileEval(, .KeyProbe, .Slots)
	}
	return 
}

func ( *hashJoinSink) ( []value.Value) bool {
	 := .keyBuf[:0]
	for ,  := range .hj.ExtSlots {
		 = value.AppendKey(, [])
	}
	.keyBuf = 
	,  := .tables[string()]
	if ! {
		 = .build()
		.tables[string()] = 
	}
	if len(.rows) == 0 {
		return true
	}
	// Cartesian probe: every build row emits per outer row (the keyless
	// disconnected join -- one bucket, the same emission protocol).
	if .hj.Cartesian {
		copy(.buf, )
		return .emitRows(, .headVal(nil), nil, 0, 0)
	}
	// Value-keyed probe: the outer row's key value looks the bucket up
	// directly -- no expand, no probe uniqueness (no relationship binds).
	// A null key never matches, per the consumed equality's semantics.
	if .keyProbe != nil {
		 := .keyProbe.Eval(.ctx, , .slots)
		if .IsNull() {
			return true
		}
		.valBuf = value.AppendKey(.valBuf[:0], )
		copy(.buf, )
		return .emitRows(, .headVal(.valBuf), nil, 0, 0)
	}
	,  := [.hj.Probe.From].AsNode()
	if ! {
		return true
	}
	// A reversed probe's target constraints belong to the outer endpoint
	// (the original hop's To); check them once per row.
	if .hj.Reversed && !.ctx.G.NodeMatcherAccepts(.probeM, ) {
		return true
	}
	copy(.buf, )
	// The candidate list keeps parallel relationships (one entry each):
	// the consumed connecting expand carries per-rel multiplicity in the
	// nested rebind, so the probe must too -- deduplicating here made the
	// extraction result-visible on multigraphs. The sort is only for
	// deterministic emission order.
	.nbuf = .ctx.G.AppendNeighborsMatched(.nbuf[:0], , .hj.Probe.Dir, .probeRM)
	slices.Sort(.nbuf)
	 := .hj.Probe.Uniq
	for ,  := range .nbuf {
		if !.hj.Reversed && !.ctx.G.NodeMatcherAccepts(.probeM, ) {
			continue
		}
		 := .headKey()
		if  < 0 {
			continue
		}
		var ,  graph.NodeID
		if  != nil {
			,  = uniqPair(.hj.Probe.Dir, , )
			if .Check && .uniq.used(.Scope, , ) {
				continue
			}
		}
		if !.emitRows(, , , , ) {
			return false
		}
	}
	return true
}

// emitRows binds each hit row's payload into the buffered outer row and
// emits it through the pair replay protocol: captured Check pairs replay
// against the outer used-pair env, captured live pairs (plus the probe
// hop's own, when there is one) push for downstream ops. Reports the
// downstream keep-going verdict; a stop still restores the pair stack.
func ( *hashJoinSink) ( *hjTable,  int32,  *plan.RelUniq, ,  graph.NodeID) bool {
	for  := ;  >= 0;  = .rows[].next {
		 := &.rows[]
		 := false
		for ,  := range .pairs {
			if .check && .uniq.used(.scope, .a, .b) {
				 = true
				break
			}
			// The probe hop's own pair against this row's live pairs: in
			// effective execution order the build ops precede the probe, so
			// a Check-marked probe must not reuse a relationship the row
			// already bound. (The outer env cannot catch this -- the row's
			// pairs push only after the probe's own check ran.)
			if  != nil && .Check && .live && .scope == .Scope && .a ==  && .b ==  {
				 = true
				break
			}
		}
		if  {
			continue
		}
		for ,  := range .hj.PayloadSlots {
			.buf[] = .vals[]
		}
		 := len(.uniq.stack)
		if  != nil && .Contribute {
			.uniq.stack = append(.uniq.stack, uniqKey{scope: .Scope, a: , b: , check: .Check})
		}
		for ,  := range .pairs {
			if .live {
				.uniq.stack = append(.uniq.stack, uniqKey{scope: .scope, a: .a, b: .b, check: .check})
			}
		}
		 := true
		for ,  := range .where {
			if !.Eval(.ctx, .buf, .slots).IsTruthy() {
				 = false
				break
			}
		}
		 := true
		if  {
			if .count != nil {
				*.count++
			}
			 = .next.push(.buf)
		}
		.uniq.stack = .uniq.stack[:]
		if ! {
			return false
		}
	}
	return true
}

func ( *hashJoinSink) () { .next.close() }

// build runs the branch chain once over a seed row carrying only the
// external slots, materializing key -> payload rows with captured pairs.
func ( *hashJoinSink) ( []value.Value) *hjTable {
	 := &hjTable{}
	 := make([]value.Value, .seg.RowWidth)
	for ,  := range .hj.ExtSlots {
		[] = []
	}
	 := &uniqEnv{capture: true}
	 := make(map[int]bool, len(.hj.ExtSlots))
	for ,  := range .hj.ExtSlots {
		[] = true
	}
	 := &hjBuildSink{t: , hj: .hj, uniq: }
	if .hj.KeyBuild != nil {
		.keyBuild = compileEval(.ctx, .hj.KeyBuild, .seg.Slots)
		.ctx, .slots = .ctx, .seg.Slots
	}
	var  rowSink = 
	for  := len(.hj.Build) - 1;  >= 0; -- {
		 = buildStageSink(.ctx, .seg, .hj.Build[], , func( int) bool { return [] }, func(int) bool { return false }, , nil, )
	}
	.push()
	.close()
	return 
}

// hjBuildSink materializes the build chain's rows: payload values plus a
// snapshot of the capture env's stack (the pairs bound along the current
// DFS path -- exactly this row's own).
type hjBuildSink struct {
	t    *hjTable
	hj   *plan.HashJoinStage
	uniq *uniqEnv
	// keyBuild is the value-keyed join's branch-side key (nil for the
	// expand-keyed form).
	keyBuild RowEval
	ctx      *eval.Ctx
	slots    map[string]int
	valBuf   []byte
}

// push always reports true: the build chain's downstream is the join
// table, not the query sink, so a stop verdict never originates here.
func ( *hjBuildSink) ( []value.Value) bool {
	var  graph.NodeID
	if .hj.Cartesian {
		// One bucket: the probe emits every row.
	} else if .keyBuild != nil {
		// A null build key can never equal any probe value: drop the row.
		 := .keyBuild.Eval(.ctx, , .slots)
		if .IsNull() {
			return true
		}
		.valBuf = value.AppendKey(.valBuf[:0], )
	} else {
		var  bool
		,  = [.hj.KeySlot].AsNode()
		if ! {
			return true
		}
	}
	 := hjRow{next: -1}
	 := len(.t.valSlab)
	for ,  := range .hj.PayloadSlots {
		.t.valSlab = append(.t.valSlab, [])
	}
	.vals = .t.valSlab[:len(.t.valSlab):len(.t.valSlab)]
	if  := len(.uniq.stack);  > 0 {
		 := len(.t.pairSlab)
		for ,  := range .uniq.stack {
			.t.pairSlab = append(.t.pairSlab, hjPair{scope: .scope, a: .a, b: .b, check: .check, live: !.dead})
		}
		.pairs = .t.pairSlab[:len(.t.pairSlab):len(.t.pairSlab)]
	}
	 := int32(len(.t.rows))
	.t.rows = append(.t.rows, )
	var  int
	switch {
	case .hj.Cartesian:
		 = .t.byVal.GetOrCreate(nil, .t.mintChain)
	case .keyBuild != nil:
		 = .t.byVal.GetOrCreate(.valBuf, .t.mintChain)
	default:
		 = .t.byKey.GetOrCreate(uint64(), .t.mintChain)
	}
	.t.link(, )
	return true
}

func ( *hjBuildSink) () {}