rindle/op/

flipped_join.rs

//! `FlippedJoin` (`ivm/flipped-join.ts`, spec `06` §3.2) — the **flipped EXISTS**
//! inner join: child-driven, batched, outputs **parent** rows that have ≥1 related
//! child.
//!
//! Where [`Join`](crate::graph::Join) is parent-driven (fetch parents, attach a
//! lazy child relationship to each), `FlippedJoin` is child-driven: it fetches
//! **all children first**, groups them by their join key, batch-fetches the
//! matching parents in one `multi_constraints` request (the source does the k-way
//! IN-fanout merge), and yields each returned parent **with its related children
//! attached** — but only parents some child pointed at. That inner-join gate (drop
//! a parent with no related child) is exactly the `where EXISTS(rel)` semantics, so
//! the builder lowers a *flipped* EXISTS condition to a bare `FlippedJoin` (no
//! separate [`Exists`](crate::op::Exists) gate).
//!
//! ## What it reuses from `Join` (the re-materialize overlay model)
//!
//! The relationship `FlippedJoin` attaches to each parent is built **identically**
//! to `Join`'s — [`Graph::attach_child_rel`](crate::graph::Graph::attach_child_rel)
//! with the same by-value [`JoinOverlay`] + `splice_join_overlay` (Primitive #5).
//! So `FlippedJoin` inherits `Join`'s proven, View-`re-materialize` overlay
//! behavior verbatim. On top of it sits the spec's `{change, position}` in-flight
//! suppress model (`#inprogressChildChange`/`#inprogressChildChangePosition`,
//! `flipped-join.ts:103`) — the flipped twin of `Graph::join_overlay_for`,
//! see [`FlippedJoin::inprogress_overlay`] and `inflight_fetch_action`. Without
//! it, a child change fanning out to SEVERAL parents (a non-unique parent join
//! key) lets every mid-fan-out maintenance fetch see the FINAL post-change world,
//! so a downstream `Take`'s bound jumps instead of cascading and it emits phantom
//! removes (`follow-ups/take-flip-bound-crossing-edit.md`, defect #2).
//!
//! ## What is `FlippedJoin`-specific
//!
//! - **fetch** (child-first): translate the request constraint parent→child, fetch
//!   all children, group by [`canonical_key`], one batched parent fetch, drop a
//!   returned parent whose key no child produced (the inner-join gate + the
//!   pass-through filter for chained flipped joins).
//! - **push child port**: the changed child is re-projected onto its parents; for
//!   each, decide whether the parent still has *another* related child (`exists`).
//!   If yes → a `Child` change (the parent stays, its relationship changed); if no →
//!   the changed child is the parent's **sole** child, so its existence flips:
//!   `Add(parent)` on a child add, `Remove(parent)` on a child remove.
//! - **push parent port**: forward the change with the relationship flipped on,
//!   **unless** the parent has no related child (inner-join drop).
//!
//! ## Fully implemented (formerly deferred)
//!
//! - **Chunked fetch** (`computed_multi.len() > chunk_size`): slices the IN-batch and
//!   node-level-merges the chunk streams via
//!   [`merge_node_streams`](crate::op::merge_node_streams) (default `chunk_size` 256;
//!   the `chunk_size` field lets a test drive it).
//! - **Nested `Change::Child` on the child port** (a grandchild change under the
//!   flipped child): carried up as a full `Change` and rebuilt per parent, exactly as
//!   [`Join`](crate::graph)'s child port — see `push_child`/`push_child_change`.
//! - **The `{change, position}` in-flight suppress model** (see above): published by
//!   `push_child_change` for the duration of the fan-out (RAII-cleared), consulted
//!   by the fetch path. (A nested `Change::Child` publishes no overlay — it never
//!   changes membership; same reduced scope as the non-flipped twin, see
//!   [`JoinOverlay::for_child_change`].)

use std::cell::{Cell, RefCell};
use std::cmp::Ordering;
use std::collections::HashSet;

use crate::change::{
    build_join_constraint, constraints_are_compatible, materialize_change, rebuild_change,
    rebuild_node, Change, Constraint, FetchRequest, JoinOverlay, MultiConstraint, Node, NodeStream,
    OutEdge, OwnedChange, OwnedNode, Port, Relationship, SourceChange,
};
use crate::graph::{Graph, NodeId, OverlayPolarity};
use crate::value::{compare_rows, ColId, OwnedRow as Row, RelId, Sort};

use super::join_util::{
    canonical_key, canonical_key_of_constraint, is_join_match, row_equals_for_compound_key,
    CanonKey,
};

/// `MULTI_CONSTRAINT_CHUNK_SIZE` (`flipped-join.ts:55`): the IN-batch is sliced
/// into windows of this many constraints, each its own `parent.fetch`, merged.
pub const DEFAULT_CHUNK_SIZE: usize = 256;

#[cfg(any(test, feature = "testkit"))]
thread_local! {
    /// Test-only override of [`DEFAULT_CHUNK_SIZE`], read at [`FlippedJoin::new`]
    /// time (the `setMultiConstraintChunkSizeForTest` seam, `flipped-join.ts:57`).
    static CHUNK_SIZE_OVERRIDE: std::cell::Cell<Option<usize>> =
        const { std::cell::Cell::new(None) };
}

/// Override `DEFAULT_CHUNK_SIZE` for every `FlippedJoin` built after this call on
/// the current thread (`None` restores the default) — the
/// `setMultiConstraintChunkSizeForTest` seam (`flipped-join.ts:57`). Lets a test or
/// fuzz lane force the chunked k-way merge path (`merge_node_streams`) on small
/// fixtures, whose distinct child keys never approach the production 256.
/// Build-time only: an already-built graph keeps the `chunk_size` it was
/// constructed with.
#[cfg(any(test, feature = "testkit"))]
pub fn set_multi_constraint_chunk_size_for_test(n: Option<usize>) {
    CHUNK_SIZE_OVERRIDE.set(n);
}

/// [`DEFAULT_CHUNK_SIZE`], unless a test override is active on this thread.
fn default_chunk_size() -> usize {
    #[cfg(any(test, feature = "testkit"))]
    {
        CHUNK_SIZE_OVERRIDE.get().unwrap_or(DEFAULT_CHUNK_SIZE)
    }
    #[cfg(not(any(test, feature = "testkit")))]
    DEFAULT_CHUNK_SIZE
}

/// How an in-flight child fan-out reshapes one parent's place in a mid-push
/// maintenance fetch (`FlippedJoin::inflight_fetch_action`).
enum InflightAction {
    /// No in-flight change touches this parent, or the change was already
    /// delivered for it and the post-change fetch is right as-is.
    Keep,
    /// The parent must not appear in this fetch at all.
    Drop,
    /// Keep the parent, with its relationship pulled back to the PRE-change view.
    KeepPre(JoinOverlay),
}

/// `FlippedJoin`: child-driven inner join (spec `06` §3.2). Two input ports
/// ([`Port::JoinParent`]/[`Port::JoinChild`]) like `Join`;
/// the **child** drives the fetch but the **parent** is the output row, so
/// `rel_slot` resolves against the parent schema and `input_schema` is the parent's.
pub struct FlippedJoin {
    /// The output side (parent rows). The attached relationship hangs here.
    pub parent: NodeId,
    /// The driving side: children are fetched first and grouped by join key.
    pub child: NodeId,
    pub parent_key: Vec<ColId>,
    pub child_key: Vec<ColId>,
    /// Relationship slot (resolved from the alias against the **parent** schema at
    /// build time, like `Join`).
    pub rel_slot: RelId,
    /// IN-batch chunking threshold (test/seam; default `DEFAULT_CHUNK_SIZE`). A
    /// batch above this needs the node-level merge (deferred — see module docs).
    pub chunk_size: usize,
    /// The single downstream edge as a port-carrying [`OutEdge`] (like
    /// `Join`): `Port::Single` to a terminal sink, or a join
    /// port when this flipped join feeds another join.
    pub output: Cell<Option<OutEdge>>,
    /// The in-flight child change being fanned out by `push_child_change` — the
    /// flipped twin of `Join::inprogress_overlay` (Zero's `#inprogressChildChange`,
    /// `flipped-join.ts:103`). Consulted by the reentrant maintenance fetch a
    /// downstream `Take` triggers mid-push: [`FlippedJoin::fetch`] splices an
    /// in-flight REMOVEd child back into the child enumeration (so parents whose
    /// SOLE child it was are enumerated at all), and `fetch_batched` gates each
    /// parent's membership + relationship content on the fan-out position via
    /// `inflight_fetch_action`. `None` outside a child-push.
    pub(crate) inprogress_overlay: RefCell<Option<JoinOverlay>>,
    /// Zero's `#inprogressChildChangePosition` (`flipped-join.ts:104`): the row of
    /// the parent currently being delivered by the fan-out. Parents at-or-before it
    /// (in the parent's effective per-query order — the fan-out order) have had the
    /// change delivered and must see the POST-change world; parents strictly after
    /// must still see PRE-change. `None` until the fan-out reaches its first
    /// parent.
    pub(crate) inprogress_position: RefCell<Option<Row>>,
}

impl FlippedJoin {
    pub fn new(
        parent: NodeId,
        child: NodeId,
        parent_key: Vec<ColId>,
        child_key: Vec<ColId>,
        rel_slot: RelId,
    ) -> FlippedJoin {
        FlippedJoin {
            parent,
            child,
            parent_key,
            child_key,
            rel_slot,
            chunk_size: default_chunk_size(),
            output: Cell::new(None),
            inprogress_overlay: RefCell::new(None),
            inprogress_position: RefCell::new(None),
        }
    }

    /// Override the IN-batch chunk size (the `setMultiConstraintChunkSizeForTest`
    /// seam, `flipped-join.ts:57`). Used by the chunking-equivalence test once the
    /// chunked fetch path lands.
    pub fn with_chunk_size(mut self, chunk_size: usize) -> FlippedJoin {
        self.chunk_size = chunk_size;
        self
    }

    // ---------------------------------------------------------------------
    // FETCH (child-driven, batched)
    // ---------------------------------------------------------------------

    /// Lazy pull (`flipped-join.ts:161`): translate the request constraint onto the
    /// child key, fetch **all** children, then `fetch_batched`.
    pub fn fetch<'g>(&'g self, g: &'g Graph, req: &FetchRequest) -> NodeStream<'g> {
        let child_constraint = self.translate_constraint(req.constraint.as_ref());
        let child_req = FetchRequest {
            constraint: child_constraint,
            ..Default::default()
        };
        let mut children: Vec<Node<'g>> = g.fetch(self.child, &child_req).collect();
        // `flipped-join.ts:187-202`: during an in-flight child REMOVE the child
        // source already shows the row gone, so a parent whose SOLE child it was
        // would not be enumerated at all — and would silently vanish from a
        // mid-push maintenance fetch even though the fan-out has not delivered its
        // remove yet. Splice the removed child back into the enumeration; the
        // per-parent disposition in `fetch_batched` re-hides it for parents the
        // fan-out has already reached. (Append instead of Zero's sorted insert:
        // `children` only seeds the IN-batch keys here — the relationship content
        // is a re-fetch thunk, not this list.)
        if let Some(ov) = self.inprogress_overlay.borrow().as_ref() {
            if let SourceChange::Remove(row) = &ov.change {
                children.push(Node::leaf(row.clone()));
            }
        }
        self.fetch_batched(g, req, children)
    }

    /// `#fetchBatched` (`flipped-join.ts:230`): build the deduped IN-batch of
    /// child-derived parent constraints + the set of parent keys that some child
    /// produced, then one batched `parent.fetch`. Each returned parent whose key a
    /// child produced is yielded with the join relationship attached; a parent whose
    /// key no child produced is dropped (the inner-join gate, and the pass-through
    /// filter for a chained flipped join's `multi_constraints`).
    fn fetch_batched<'g>(
        &'g self,
        g: &'g Graph,
        req: &FetchRequest,
        children: Vec<Node<'g>>,
    ) -> NodeStream<'g> {
        let mut seen: HashSet<CanonKey> = HashSet::with_capacity(children.len());
        let mut computed_multi: MultiConstraint = Vec::new();
        for child in &children {
            let c = match build_join_constraint(&child.row, &self.child_key, &self.parent_key) {
                Some(c) => c,
                None => continue, // null child key cannot join
            };
            // Drop a child-derived constraint that contradicts the incoming request
            // constraint — it could never match, so it must not enter the batch.
            if let Some(rc) = req.constraint.as_ref() {
                if !constraints_are_compatible(&c, rc) {
                    continue;
                }
            }
            let k = canonical_key_of_constraint(&c);
            if seen.insert(k) {
                // First sight of this key → one IN-batch entry (dedup).
                computed_multi.push(c);
            }
        }
        if computed_multi.is_empty() {
            return Box::new(std::iter::empty());
        }

        // Build the parent stream: one batched fetch when the IN-batch fits a single
        // chunk, else one fetch per `chunk_size` window, k-way-merged in parent order
        // (`#fetchChunked`, `flipped-join.ts:311`). The windows are disjoint key-sets
        // so the merge does NOT dedup. The source turns each request's
        // `multi_constraints` into the per-entry IN-fanout + merge.
        let parent_stream: NodeStream<'g> = if computed_multi.len() <= self.chunk_size {
            g.fetch(self.parent, &self.parent_batch_req(req, computed_multi))
        } else {
            // Each chunk fetch streams in the parent's **resolved per-query order**
            // (`input_sort` — e.g. `order by Milliseconds`), NOT the base table's PK
            // sort (`input_schema().sort`). The k-way merge must compare on that same
            // resolved order or it interleaves the chunks wrongly and a downstream
            // `Take` keeps the wrong top-N. (Only bites above `chunk_size` distinct
            // child keys, so the mini fixtures never hit it — the chinook scale sweep
            // did.) See `input_sort` vs `input_schema` in `graph.rs`.
            let parent_sort = g.input_sort(self.parent);
            let chunks: Vec<NodeStream<'g>> = computed_multi
                .chunks(self.chunk_size)
                .map(|window| g.fetch(self.parent, &self.parent_batch_req(req, window.to_vec())))
                .collect();
            crate::op::merge_node_streams(chunks, parent_sort, req.reverse, None)
        };

        // Capture by value for the `'g` filter_map closure (`self` is `&'g` too —
        // the in-flight disposition below re-reads the live fields through it).
        let parent_key = self.parent_key.clone();
        let child_key = self.child_key.clone();
        let child_id = self.child;
        let rel_slot = self.rel_slot;
        // The fan-out position gate compares in the parent's effective per-query
        // order; resolving it walks (and allocates) the operator chain, so it is
        // cached once per fetch here, not recomputed per yielded parent.
        let mut parent_sort: Option<Sort> = None;
        Box::new(parent_stream.filter_map(move |pnode| {
            let k = canonical_key(&pnode.row, &parent_key);
            // Miss ⇒ no child produced this parent's key ⇒ drop it (inner-join gate
            // / chained-flip pass-through filter, `flipped-join.ts:295`).
            if !seen.contains(&k) {
                return None;
            }
            // While a child change is being fanned out, membership and relationship
            // content are position-dependent (Zero `#yieldParentWithOverlay`,
            // `flipped-join.ts:332`) — the flipped twin of `Graph::join_overlay_for`.
            let overlay = match self.inflight_fetch_action(g, &pnode.row, &mut parent_sort) {
                InflightAction::Keep => None,
                InflightAction::Drop => return None,
                InflightAction::KeepPre(ov) => Some((ov, OverlayPolarity::Pre)),
            };
            Some(g.attach_child_rel(child_id, &parent_key, &child_key, rel_slot, pnode, overlay))
        }))
    }

    /// Decide, at yield time, how the in-flight child change reshapes `parent_row`'s
    /// place in a mid-push maintenance fetch — the flipped twin of
    /// `Graph::join_overlay_for`, porting Zero's `#yieldParentWithOverlay`
    /// (`flipped-join.ts:332-383`) to the re-fetch relationship model.
    ///
    /// A parent at-or-before the fan-out position (in the parent's effective
    /// per-query order — the order `push_child_change` delivers) has had the change
    /// delivered: the post-change world is correct, but an in-flight REMOVE needs
    /// its membership re-checked (the removed child was spliced back into the
    /// enumeration by `fetch`, so a sole-child parent would otherwise reappear). A
    /// parent strictly after the position must still see the PRE-change world:
    /// membership (an in-flight ADD that is its only child does not exist for it
    /// yet) and relationship content (the `Pre`-polarity overlay splice).
    fn inflight_fetch_action(
        &self,
        g: &Graph,
        parent_row: &Row,
        parent_sort: &mut Option<Sort>,
    ) -> InflightAction {
        // Gate on a BORROWED overlay: the common outcomes (no in-flight change, an
        // unrelated parent, already-delivered) must not pay the two-row overlay
        // clone — only the `KeepPre` arms clone. Holding the `Ref` across the
        // fetches below is fine: fetch paths only ever `borrow()` these fields;
        // the sole `borrow_mut` writers are in `push_child_change`, outside any
        // fetch.
        let ov_ref = self.inprogress_overlay.borrow();
        let Some(ov) = ov_ref.as_ref() else {
            return InflightAction::Keep;
        };
        if !is_join_match(
            parent_row,
            &self.parent_key,
            ov.change.row(),
            &self.child_key,
        ) {
            return InflightAction::Keep;
        }
        let delivered = match self.inprogress_position.borrow().as_ref() {
            // Same comparator note as `Graph::join_overlay_for`: the fan-out
            // delivers parents in the parent's EFFECTIVE per-query order, so the
            // "already delivered?" gate must compare in that order (resolved once
            // per fetch — the caller-held cache).
            Some(pos) => {
                let sort = parent_sort.get_or_insert_with(|| g.input_sort(self.parent));
                compare_rows(sort, parent_row, pos) != Ordering::Greater
            }
            // The fan-out has not reached its first parent — nothing delivered.
            None => false,
        };
        match (&ov.change, delivered) {
            // Delivered REMOVE: post-change is right, but `fetch` spliced the
            // removed child back into the enumeration — re-gate membership on the
            // post-change truth so a sole-child parent stays gone.
            (SourceChange::Remove(_), true) => {
                if self.has_related_child(g, parent_row) {
                    InflightAction::Keep
                } else {
                    InflightAction::Drop
                }
            }
            // Pending REMOVE: pre-change world — the child is still related (the
            // splice keeps it in the enumeration; the Pre overlay re-adds it to the
            // relationship refetch).
            (SourceChange::Remove(_), false) => InflightAction::KeepPre(ov.clone()),
            // Pending ADD: pre-change world — the added child does not exist yet
            // for this parent; if it is the parent's ONLY child, the parent was not
            // in the pre-change output at all.
            (SourceChange::Add(row), false) => {
                let child_pk = g.input_schema(self.child).primary_key.clone();
                if self.parent_has_other_child(g, parent_row, row, &child_pk) {
                    InflightAction::KeepPre(ov.clone())
                } else {
                    InflightAction::Drop
                }
            }
            // Delivered ADD: the post-change fetch (child present) is correct.
            (SourceChange::Add(_), true) => InflightAction::Keep,
            // EDIT never changes membership (the join key is immutable, asserted on
            // entry); a pending parent still sees the OLD child content.
            (SourceChange::Edit { .. }, false) => InflightAction::KeepPre(ov.clone()),
            (SourceChange::Edit { .. }, true) => InflightAction::Keep,
        }
    }

    /// Build the parent fetch request for one IN-batch (the whole `computed_multi`
    /// or a single chunk window): the incoming constraint ANDed with
    /// `[...incoming_multi, batch]` (a chained flipped join above prepends its own
    /// `multi_constraints`), carrying the request's `start`/`reverse`.
    fn parent_batch_req(&self, req: &FetchRequest, batch: MultiConstraint) -> FetchRequest {
        let mut multi = req.multi_constraints.clone();
        multi.push(batch);
        FetchRequest {
            constraint: req.constraint.clone(),
            multi_constraints: multi,
            start: req.start.clone(),
            reverse: req.reverse,
        }
    }

    /// Translate the parent-key columns of an incoming request constraint to the
    /// corresponding child-key columns (`flipped-join.ts:164`). Columns that aren't
    /// parent-key columns are dropped; if none translate, the child is fetched
    /// unconstrained (`None`).
    fn translate_constraint(&self, c: Option<&Constraint>) -> Option<Constraint> {
        let c = c?;
        let mut out = Constraint::new();
        for (col, val) in c {
            if let Some(i) = self.parent_key.iter().position(|pk| pk == col) {
                out.push((self.child_key[i], val.clone()));
            }
        }
        if out.is_empty() {
            None
        } else {
            Some(out)
        }
    }

    // ---------------------------------------------------------------------
    // PUSH (two ports)
    // ---------------------------------------------------------------------

    pub fn push<'g>(&'g self, g: &'g Graph, change: Change<'g>, port: Port) {
        match port {
            Port::JoinChild => self.push_child(g, change),
            Port::JoinParent => self.push_parent(g, change),
            Port::Single => panic!("FlippedJoin received Single port"),
        }
    }

    /// Child port (`flipped-join.ts:385`): the changed child is re-projected onto
    /// its parents with the existence decision. EDIT/CHILD can't flip a parent's
    /// existence, so they pre-set `exists = true`. A key edit is rejected (the
    /// source split it, like [`Join`](crate::graph)).
    fn push_child<'g>(&'g self, g: &'g Graph, change: Change<'g>) {
        let exists_pre = match &change {
            Change::Add(_) | Change::Remove(_) => false,
            Change::Edit { node, old } => {
                assert!(
                    row_equals_for_compound_key(&old.row, &node.row, &self.child_key),
                    "flipped child edit must not change the join relationship key"
                );
                // A non-key edit cannot make a parent appear/disappear → forward a
                // Child change (existence is unchanged).
                true
            }
            // A nested `Change::Child` (a grandchild change) leaves the direct child
            // present, so the parent's EXISTS-membership is unchanged → forward it as
            // a `Child`, carried verbatim. `exists_pre = true` skips the membership
            // re-check; without it, `parent_has_other_child` (which excludes the
            // *changed* child) could see "no other child" and wrongly emit an
            // Add/Remove of the parent for what is only a grandchild edit.
            Change::Child { .. } => true,
        };
        self.push_child_change(g, change, exists_pre);
    }

    /// `#pushChildChange` (`flipped-join.ts:409`), the reentrant crux. Build the
    /// parent constraint from the changed child, re-enter `parent.fetch`, and for
    /// each parent decide existence (does it have *another* related child?). Yes →
    /// `Child`; no → the child is the parent's sole child, so the parent's existence
    /// flips (`Add` on child-add / `Remove` on child-remove).
    fn push_child_change<'g>(&'g self, g: &'g Graph, child_change: Change<'g>, exists_pre: bool) {
        let out = self.output.get().expect("FlippedJoin output not wired");
        let key_row = child_change.primary_row().clone();
        let constraint = match build_join_constraint(&key_row, &self.child_key, &self.parent_key) {
            Some(c) => c,
            None => return, // null child key cannot join
        };
        let overlay = JoinOverlay::for_child_change(&child_change);
        let child_pk = g.input_schema(self.child).primary_key.clone();
        let is_add = matches!(child_change, Change::Add(_));
        // Materialize once (drains thunks now), rebuild a fresh `Change<'g>` per
        // parent in the `exists` branch — a child can join to many parents.
        let owned = materialize_change(child_change);

        // Publish the in-flight child change LIVE for the duration of the fan-out
        // (Zero `flipped-join.ts:410`); the guard clears both fields on every exit
        // (the JS `try/finally`, `flipped-join.ts:485` — under `panic = "abort"` a
        // panic aborts instead and `Drop` is skipped, see WS02).
        *self.inprogress_overlay.borrow_mut() = overlay.clone();
        let _guard = InprogressGuard { j: self };

        // Collect the reentrant parent fetch (relationships preserved) so its stream
        // borrow is dropped before the per-parent existence fetch + downstream push.
        let parents: Vec<Node> = g
            .fetch(self.parent, &FetchRequest::with_constraint(constraint))
            .collect();

        // `exists` is sticky across the parent loop (the JS captures it in method
        // scope, `flipped-join.ts:436`): once any parent has another child, later
        // parents stay in the `Child` branch. Moot in the common 1:1 case.
        let mut exists = exists_pre;
        // The flipped twin of the unbounded join fan-out (RUNAWAY-PUSH-FINDINGS §3), with a
        // per-parent existence fetch on top — deadline-checkpointed like the non-flipped loop
        // (FOLLOWER-LAG-SHED §6.6).
        for pnode in parents {
            if g.push_deadline_exceeded() {
                g.park_runtime_error(crate::error::RindleError::PushDeadlineExceeded {
                    site: "flipped-join child fan-out",
                });
                break; // torn is fine — the host discards the engine
            }
            // Advance the fan-out position to the parent now being delivered
            // (`flipped-join.ts:427`) BEFORE the downstream push: a maintenance
            // fetch it triggers must see this parent as delivered, and every
            // later parent as still pre-change.
            *self.inprogress_position.borrow_mut() = Some(pnode.row.clone());
            if !exists {
                exists = self.parent_has_other_child(g, &pnode.row, &key_row, &child_pk);
            }
            if exists {
                // Parent stays; its relationship changed → a Child change. (Both the
                // View and Catch read only the parent row on a `Child` and recurse
                // into the carried sub-change; the overlay is `None` for a nested
                // Child and a leaf-row splice otherwise.)
                let pn = g.attach_child_rel(
                    self.child,
                    &self.parent_key,
                    &self.child_key,
                    self.rel_slot,
                    pnode,
                    overlay.clone().map(|ov| (ov, OverlayPolarity::Post)),
                );
                let child = Box::new(rebuild_change(owned.clone()));
                g.push(
                    out.node,
                    Change::Child {
                        node: pn,
                        rel: self.rel_slot,
                        child,
                    },
                    out.port,
                );
            } else {
                // The changed child is the parent's SOLE related child → existence
                // flips. The relationship is exactly that one child (JS `[change[NODE]]`,
                // `flipped-join.ts:474`) — NOT a re-fetch (which, on a remove, would
                // show the child already gone). This makes the emitted Add/Remove
                // carry the changed child, matching the `Exists`-gate oracle.
                //
                // `[change[NODE]]` is the change's NODE — the child *with its own
                // relationships* — not its row. Passing the bare row here truncated the
                // child's own witness subtree, which the `View`, the cluster snapshot and
                // the d2s oracle all discard anyway (none projects a `where`-EXISTS slot)
                // but the normalized wire fold refcounts in full: a nested EXISTS witness
                // dropped here never enters the client's footprint, so its re-derivation
                // fails the inner gate and drops the root. `owned` is the same
                // materialization the `Child` arm above forwards.
                let child_owned: OwnedNode = match &owned {
                    OwnedChange::Add(n) | OwnedChange::Remove(n) => n.clone(),
                    // Unreachable: `push_child` pre-sets `exists = true` for both, so
                    // they never reach the flip branch. Matched for totality.
                    OwnedChange::Edit { node, .. } | OwnedChange::Child { node, .. } => {
                        node.clone()
                    }
                };
                let pn = self.parent_with_only_child(pnode, child_owned);
                let flipped = if is_add {
                    Change::Add(pn)
                } else {
                    Change::Remove(pn)
                };
                g.push(out.node, flipped, out.port);
            }
        }
    }

    /// Does `parent_row` have a related child whose PK differs from the changed
    /// child (`change.node`)? Reentrant `child.fetch` (the child source overlay is
    /// active, so a child add is visible / a child remove is gone), then look for any
    /// child that is **not** the changed one — a PK match (the robust port of the JS
    /// `compareRows(child, change.node) !== 0`, `flipped-join.ts:441`; spec OQ-7).
    fn parent_has_other_child(
        &self,
        g: &Graph,
        parent_row: &Row,
        changed_child: &Row,
        child_pk: &[ColId],
    ) -> bool {
        let Some(cc) = build_join_constraint(parent_row, &self.parent_key, &self.child_key) else {
            return false;
        };
        for cn in g.fetch(self.child, &FetchRequest::with_constraint(cc)) {
            if !row_equals_for_compound_key(&cn.row, changed_child, child_pk) {
                return true;
            }
        }
        false
    }

    /// Parent port (`flipped-join.ts:490`): forward the change with the relationship
    /// flipped on — **unless** the parent has no related child, in which case it is
    /// not in the (inner-join) output and the change is dropped.
    fn push_parent<'g>(&'g self, g: &'g Graph, change: Change<'g>) {
        let out = self.output.get().expect("FlippedJoin output not wired");
        match change {
            Change::Add(node) => {
                if !self.has_related_child(g, &node.row) {
                    return;
                }
                let pn = self.flip(g, node);
                g.push(out.node, Change::Add(pn), out.port);
            }
            Change::Remove(node) => {
                if !self.has_related_child(g, &node.row) {
                    return;
                }
                let pn = self.flip(g, node);
                g.push(out.node, Change::Remove(pn), out.port);
            }
            Change::Edit { node, old } => {
                assert!(
                    row_equals_for_compound_key(&old.row, &node.row, &self.parent_key),
                    "flipped parent edit must not change the join key"
                );
                // Key unchanged ⇒ old and new have the same related children; gate on
                // either. No children ⇒ the parent isn't in the output ⇒ drop.
                if !self.has_related_child(g, &node.row) {
                    return;
                }
                let pn = self.flip(g, node);
                let po = self.flip(g, old);
                g.push(out.node, Change::Edit { node: pn, old: po }, out.port);
            }
            Change::Child { node, rel, child } => {
                // Passthrough of a relationship above this flipped join.
                if !self.has_related_child(g, &node.row) {
                    return;
                }
                let pn = self.flip(g, node);
                g.push(
                    out.node,
                    Change::Child {
                        node: pn,
                        rel,
                        child,
                    },
                    out.port,
                );
            }
        }
    }

    /// The inner-join gate: does `parent_row` have ≥1 related child? (`child.fetch`
    /// of the correlation constraint has a first row.)
    fn has_related_child(&self, g: &Graph, parent_row: &Row) -> bool {
        match build_join_constraint(parent_row, &self.parent_key, &self.child_key) {
            Some(c) => g
                .fetch(self.child, &FetchRequest::with_constraint(c))
                .next()
                .is_some(),
            None => false,
        }
    }

    /// Attach the join's child relationship to a parent node (the re-materialize
    /// thunk, no overlay — the parent-port push has no in-flight child of its own).
    fn flip<'g>(&self, g: &'g Graph, node: Node<'g>) -> Node<'g> {
        g.attach_child_rel(
            self.child,
            &self.parent_key,
            &self.child_key,
            self.rel_slot,
            node,
            None,
        )
    }

    /// Attach a relationship that yields exactly the one given child — the parent's
    /// sole related child on an existence flip (JS `[change[NODE]]`). The child is an
    /// [`OwnedNode`], not a row: it carries its own relationships (a nested EXISTS
    /// witness, a `related` subtree), and every consumer that walks the whole subtree
    /// — the normalized wire fold above all — needs them.
    fn parent_with_only_child<'g>(&self, mut parent: Node<'g>, child: OwnedNode) -> Node<'g> {
        parent.rels.push(Relationship {
            slot: self.rel_slot,
            thunk: Box::new(move || Box::new(std::iter::once(rebuild_node(child.clone())))),
        });
        parent
    }
}

/// RAII for the flipped join's in-flight child fan-out (`flipped-join.ts:485-487`
/// `finally`): clears [`FlippedJoin::inprogress_overlay`]/
/// [`FlippedJoin::inprogress_position`] when `push_child_change` returns —
/// normally, via `?`, or unwinding (the latter only under `panic = "unwind"`;
/// under the shipping `panic = "abort"` profile a panic aborts and `Drop` is
/// skipped — see WS02). A stale overlay would corrupt every later fetch's
/// membership, so the clear is not optional. The twin of `graph.rs`'s
/// `InprogressGuard` for the non-flipped `Join`.
struct InprogressGuard<'a> {
    j: &'a FlippedJoin,
}

impl Drop for InprogressGuard<'_> {
    fn drop(&mut self) {
        *self.j.inprogress_overlay.borrow_mut() = None;
        *self.j.inprogress_position.borrow_mut() = None;
    }
}