rindle_wire/

family_key.rs

//! [`FamilyKey`] — the identity of a **parameterized query family** (design 310 §3):
//! the [`QueryKey`](crate::query_key::QueryKey) construction over a query's *template*,
//! the canonical AST with its eligible root-equality literals punched into holes. Two
//! subscriptions are family-mates iff their `FamilyKey`s are equal; the extracted
//! literal tuple — the [`Binding`], a [`CanonKey`] — is what distinguishes them.
//!
//! Extraction is pure data work: no schema, no builder, no engine. What it decides:
//!
//! - **Which literals are holes** (design §3.1): the RHS of every root-level equality
//!   conjunct — the single top-level `Simple`, or the direct `Simple` children of a
//!   top-level `And` — whose op is `=`, whose LHS is a column, and whose literal is a
//!   scalar a family may bind on ([`canon_of_lit`]: not `Null`, not an array). *Every*
//!   eligible conjunct is holed, no inference: two ASTs group iff they agree everywhere
//!   else, and a conjunct whose literal happens to agree across bindings simply
//!   contributes a constant column to the binding tuple.
//! - **What the key hashes** (impl plan D2): the canonical JSON of the
//!   number-canonicalized AST with each holed conjunct's `right` replaced **in position**
//!   by `{"hole": i}`. Position is kept deliberately: conjunct order participates in the
//!   template (design §3.1 — there is no `normalize_ast` pass), so `And[a=?, b>3]` and
//!   `And[b>3, a=?]` are different families. A key over the stripped AST alone would
//!   silently widen grouping.
//! - **What the builder compiles** ([`FamilyTemplate::stripped`]): the same canonical
//!   AST with the holed conjuncts *removed* (an `And` left with one child collapses to
//!   that child; zero ⇒ `where: None`), so the engine's family root connection carries
//!   only the residual predicate and the membership test replaces the holes.
//! - **What does not group** (design §3.2): an aggregate root (`aggregate` /
//!   `group_by` / `having`) is refused outright; a query with no eligible conjunct
//!   yields `Ok(None)` and falls back to today's per-query materialization, byte for
//!   byte. Everything outside the holes — `related` structure, subquery literals,
//!   `order_by`, `limit`, `one`, `select`, `start` — stays in the key, so it separates
//!   families exactly as it separates `QueryKey`s.
//!
//! One shape the engine's alias normalization would otherwise reject is also refused
//! here: when stripping leaves the root `where` a **bare** EXISTS whose alias collides
//! with a materialized `related` of the same name. The concrete query's `And` gets that
//! alias uniquified (`x` → `x_0`); the stripped template's singleton does not, and the
//! builder refuses the collision. Such a query stays a singleton.

use std::fmt;
use std::hash::{Hash, Hasher};

use rindle::canon::{CanonKey, CanonVal};
use rindle::{
    canon_of_lit, canonicalize_wire_number_lits, Ast, Condition, Lit, Op, SimpleCondition,
    ValuePosition,
};
use serde_json::{json, Value};

use crate::query_key::{
    hash_field, normalize_select_arrays, write_canonical_json, Fnv1a64, QueryKeyError, StreamMode,
};

/// A family member's parameter tuple: the canonical literal of each holed conjunct, in
/// hole order. A [`CanonKey`] (impl plan D1) — the *same* type the engine's partition
/// membership test and the daemon's per-partition demux key by, so the three agree by
/// construction. `Send + Sync` (the payloads are `Arc<str>`), so it crosses worker
/// command channels.
pub type Binding = CanonKey;

/// The compiled half of a family: the stripped AST the builder lowers, plus the names
/// of the holed columns (the partition key, resolved to `ColId`s by the builder) and
/// each hole's position among the original top-level conjuncts (what
/// [`instantiate`](FamilyTemplate::instantiate) needs to be an exact inverse).
#[derive(Clone, Debug, PartialEq)]
pub struct FamilyTemplate {
    /// The number-canonicalized AST with the holed conjuncts removed.
    pub stripped: Ast,
    /// The holed conjuncts' column names, in hole order.
    pub params: Vec<Box<str>>,
    /// Each hole's index among the concrete query's top-level conjuncts (`0` for a
    /// query whose whole `where` was the one holed conjunct).
    pub positions: Vec<usize>,
}

impl FamilyTemplate {
    /// The concrete AST for one binding — the exact inverse of extraction on the
    /// number-canonicalized AST (`instantiate(extract(a).template, extract(a).binding)
    /// == canonicalized(a)`). Used by the engine's differential (the standalone twin
    /// of a partition), by the daemon's per-partition normalize fold, and by
    /// migration.
    pub fn instantiate(&self, binding: &Binding) -> Ast {
        assert_eq!(
            binding.len(),
            self.params.len(),
            "binding arity must match the template's parameter count"
        );
        let residual: Vec<Condition> = match &self.stripped.r#where {
            None => Vec::new(),
            Some(Condition::And { conditions }) => conditions.clone(),
            Some(other) => vec![other.clone()],
        };
        let total = residual.len() + self.params.len();
        let mut residual = residual.into_iter();
        let mut conjuncts: Vec<Condition> = Vec::with_capacity(total);
        for i in 0..total {
            match self.positions.iter().position(|&p| p == i) {
                Some(k) => conjuncts.push(Condition::Simple(SimpleCondition {
                    op: Op::Eq,
                    left: ValuePosition::Column {
                        name: self.params[k].clone(),
                    },
                    right: ValuePosition::Literal {
                        value: lit_of_canon(&binding[k]),
                    },
                })),
                None => conjuncts.push(
                    residual
                        .next()
                        .expect("template positions are consistent with its residual"),
                ),
            }
        }
        let mut ast = self.stripped.clone();
        ast.r#where = match conjuncts.len() {
            0 => None,
            1 => conjuncts.pop(),
            _ => Some(Condition::And {
                conditions: conjuncts,
            }),
        };
        ast
    }
}

/// The literal spelling of a binding value: the wire-token identity
/// `canonicalize_wire_number_lits` would produce (`Int` for the integral class, the
/// exact bits otherwise). `Null`/`Absent`/`Json` never come out of extraction
/// ([`canon_of_lit`] refuses them); they are mapped for totality only.
fn lit_of_canon(v: &CanonVal) -> Lit {
    match v {
        CanonVal::Absent | CanonVal::Null => Lit::Null,
        CanonVal::Bool(b) => Lit::Bool(*b),
        CanonVal::Int(i) => Lit::Int(*i),
        CanonVal::Float(bits) => Lit::Number(f64::from_bits(*bits)),
        CanonVal::Str(s) | CanonVal::Json(s) => Lit::Str(Box::from(&**s)),
    }
}

/// Every daemon-side input that can affect the emitted rows of a family: the same four
/// fields as [`QueryKey`](crate::query_key::QueryKey), with the canonical **template**
/// bytes in place of the canonical AST bytes.
#[derive(Clone, Eq)]
pub struct FamilyKey {
    schema_version: String,
    stream_mode: StreamMode,
    visibility_key: String,
    canonical_template: Vec<u8>,
    fingerprint: u64,
}

impl FamilyKey {
    pub fn schema_version(&self) -> &str {
        &self.schema_version
    }

    pub fn stream_mode(&self) -> StreamMode {
        self.stream_mode
    }

    pub fn visibility_key(&self) -> &str {
        &self.visibility_key
    }

    /// The canonical template bytes: the canonical AST JSON with `{"hole": i}` in each
    /// holed conjunct's `right` position.
    pub fn canonical_template(&self) -> &[u8] {
        &self.canonical_template
    }

    pub fn fingerprint(&self) -> u64 {
        self.fingerprint
    }

    pub fn fingerprint_hex(&self) -> String {
        format!("{:016x}", self.fingerprint)
    }
}

impl fmt::Debug for FamilyKey {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("FamilyKey")
            .field("schema_version", &self.schema_version)
            .field("stream_mode", &self.stream_mode)
            .field("visibility_key", &self.visibility_key)
            .field(
                "canonical_template",
                &String::from_utf8_lossy(&self.canonical_template),
            )
            .field("fingerprint", &self.fingerprint_hex())
            .finish()
    }
}

impl PartialEq for FamilyKey {
    fn eq(&self, other: &Self) -> bool {
        self.schema_version == other.schema_version
            && self.stream_mode == other.stream_mode
            && self.visibility_key == other.visibility_key
            && self.canonical_template == other.canonical_template
    }
}

impl Hash for FamilyKey {
    fn hash<H: Hasher>(&self, state: &mut H) {
        self.schema_version.hash(state);
        self.stream_mode.hash(state);
        self.visibility_key.hash(state);
        self.canonical_template.hash(state);
    }
}

/// The result of a successful extraction: the family the query belongs to, the template
/// its pipeline compiles from, and the binding this query is.
#[derive(Clone, Debug)]
pub struct FamilyExtraction {
    pub key: FamilyKey,
    pub template: FamilyTemplate,
    pub binding: Binding,
}

/// Extract `ast`'s family, if it can join one (module docs). `Ok(None)` ⇒ the query
/// cannot join any family and falls back to its `QueryKey` — today's path, byte for
/// byte. `Err` only on the (practically unreachable) serialization failure `QueryKey`
/// shares.
pub fn extract_family(
    schema_version: impl Into<String>,
    stream_mode: StreamMode,
    ast: &Ast,
    visibility_key: impl Into<String>,
) -> Result<Option<FamilyExtraction>, QueryKeyError> {
    // Design §3.2: aggregates are v1-excluded outright (the builder dispatches them to
    // a different lowering the family entry does not touch).
    if ast.aggregate.is_some() || !ast.group_by.is_empty() || ast.having.is_some() {
        return Ok(None);
    }
    // The wire-token identity first (impl plan F2 / D2): `Int(1)` and `Number(1.0)`
    // must be one binding, and the key must not fork on a spelling.
    let mut canon = ast.clone();
    canonicalize_wire_number_lits(&mut canon);

    let Some(root) = canon.r#where.as_ref() else {
        return Ok(None);
    };
    // (position, column, canonical literal) of every hole, in conjunct order.
    let mut holes: Vec<(usize, Box<str>, CanonVal)> = Vec::new();
    let mut residual: Vec<Condition> = Vec::new();
    match root {
        Condition::Simple(sc) => match hole_of(sc) {
            Some((col, v)) => holes.push((0, col, v)),
            None => return Ok(None),
        },
        Condition::And { conditions } => {
            for (i, c) in conditions.iter().enumerate() {
                match c {
                    Condition::Simple(sc) => match hole_of(sc) {
                        Some((col, v)) => holes.push((i, col, v)),
                        None => residual.push(c.clone()),
                    },
                    other => residual.push(other.clone()),
                }
            }
            if holes.is_empty() {
                return Ok(None);
            }
        }
        // A root `Or` or a bare EXISTS holds no eligible conjunct.
        Condition::Or { .. } | Condition::CorrelatedSubquery(_) => return Ok(None),
    }

    // The stripped template.
    let residual_was_one = residual.len() == 1;
    let stripped_where = match residual.len() {
        0 => None,
        1 => residual.pop(),
        _ => Some(Condition::And {
            conditions: residual,
        }),
    };
    if bare_exists_alias_collides(stripped_where.as_ref(), &canon) {
        return Ok(None);
    }
    // A lone residual that is itself an `And` stays wrapped in a one-conjunct `And`:
    // left bare, [`FamilyTemplate::instantiate`] would read its conjuncts as several
    // residuals and rebuild a flatter tree than the concrete query — `instantiate ∘
    // extract` would not be the identity (found by the fuzz corpus round-trip,
    // `rindle-fuzz/tests/family_key_roundtrip.rs`). The alias-collision check above runs
    // on the unwrapped residual, exactly as before.
    let stripped_where = match stripped_where {
        Some(inner @ Condition::And { .. }) if residual_was_one => Some(Condition::And {
            conditions: vec![inner],
        }),
        other => other,
    };
    let mut stripped = canon.clone();
    stripped.r#where = stripped_where;

    // The key: canonical JSON with holes punched in place.
    let mut value = serde_json::to_value(&canon).map_err(QueryKeyError)?;
    normalize_select_arrays(&mut value);
    punch_holes(&mut value, &holes);
    let mut canonical_template = Vec::new();
    write_canonical_json(&value, &mut canonical_template)?;

    let schema_version = schema_version.into();
    let visibility_key = visibility_key.into();
    let mut hasher = Fnv1a64::new();
    // Domain-separate from `QueryKey` so a family fingerprint never collides with a
    // query fingerprint by construction, even though they key different maps.
    hash_field(&mut hasher, b"family");
    hash_field(&mut hasher, schema_version.as_bytes());
    hash_field(&mut hasher, stream_mode.as_str().as_bytes());
    hash_field(&mut hasher, visibility_key.as_bytes());
    hash_field(&mut hasher, &canonical_template);
    let fingerprint = hasher.finish();

    let positions = holes.iter().map(|(p, _, _)| *p).collect();
    let params = holes.iter().map(|(_, c, _)| c.clone()).collect();
    let binding: Binding = holes.into_iter().map(|(_, _, v)| v).collect();
    Ok(Some(FamilyExtraction {
        key: FamilyKey {
            schema_version,
            stream_mode,
            visibility_key,
            canonical_template,
            fingerprint,
        },
        template: FamilyTemplate {
            stripped,
            params,
            positions,
        },
        binding,
    }))
}

/// The hole test (design §3.1): `Column = <bindable scalar literal>`.
fn hole_of(sc: &SimpleCondition) -> Option<(Box<str>, CanonVal)> {
    if sc.op != Op::Eq {
        return None;
    }
    let ValuePosition::Column { name } = &sc.left else {
        return None;
    };
    let ValuePosition::Literal { value } = &sc.right else {
        return None;
    };
    canon_of_lit(value).map(|v| (name.clone(), v))
}

/// Would the builder refuse the stripped template where it accepts the concrete
/// query? The one such shape: stripping leaves a *bare* EXISTS (the singleton
/// `normalize_pipeline_ast` does not alias-uniquify) whose alias — `""` when absent,
/// exactly as the builder spells it — names a materialized `related` of the same
/// query. A `limit 0` EXISTS builds no join and claims no slot, so it cannot collide.
fn bare_exists_alias_collides(stripped_where: Option<&Condition>, ast: &Ast) -> bool {
    fn bare(cond: &Condition) -> Option<&rindle::CorrelatedSubqueryCondition> {
        match cond {
            Condition::CorrelatedSubquery(c) => Some(c),
            Condition::And { conditions } | Condition::Or { conditions }
                if conditions.len() == 1 =>
            {
                bare(&conditions[0])
            }
            _ => None,
        }
    }
    let Some(c) = stripped_where.and_then(bare) else {
        return false;
    };
    if c.related.subquery.limit == Some(0) {
        return false;
    }
    let alias = c.related.subquery.alias.as_deref().unwrap_or("");
    ast.related
        .iter()
        .filter_map(|r| r.subquery.alias.as_deref())
        .any(|a| a == alias)
}

/// Replace each holed conjunct's `right` in the AST JSON with `{"hole": k}`. The walk
/// mirrors the eligibility walk exactly: `where` is either the single holed `simple`
/// or an `and` whose `conditions[i]` are the holes.
fn punch_holes(value: &mut Value, holes: &[(usize, Box<str>, CanonVal)]) {
    let Some(w) = value.get_mut("where") else {
        return;
    };
    let is_and = w.get("type").and_then(Value::as_str) == Some("and");
    for (k, (pos, _, _)) in holes.iter().enumerate() {
        let target = if is_and {
            w.get_mut("conditions")
                .and_then(Value::as_array_mut)
                .and_then(|cs| cs.get_mut(*pos))
        } else {
            debug_assert_eq!(*pos, 0);
            Some(&mut *w)
        };
        if let Some(t) = target {
            t["right"] = json!({ "hole": k });
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::query_key::QueryKey;
    use rindle::{
        table, Bound, CorrelatedSubquery, CorrelatedSubqueryCondition, Correlation, Dir, ExistsOp,
        OrderPart,
    };
    use std::collections::BTreeMap;

    fn extract(ast: &Ast) -> Option<FamilyExtraction> {
        extract_family("v1", StreamMode::Normalized, ast, "vis").expect("extract")
    }
    fn key(ast: &Ast) -> FamilyKey {
        extract(ast).expect("eligible").key
    }
    fn binding(ast: &Ast) -> Binding {
        extract(ast).expect("eligible").binding
    }
    fn col(name: &str) -> ValuePosition {
        ValuePosition::Column { name: name.into() }
    }
    fn lit(v: Lit) -> ValuePosition {
        ValuePosition::Literal { value: v }
    }
    fn simple(name: &str, op: Op, v: Lit) -> Condition {
        Condition::Simple(SimpleCondition {
            op,
            left: col(name),
            right: lit(v),
        })
    }
    fn eq(name: &str, v: Lit) -> Condition {
        simple(name, Op::Eq, v)
    }
    fn and(conditions: Vec<Condition>) -> Condition {
        Condition::And { conditions }
    }
    fn or(conditions: Vec<Condition>) -> Condition {
        Condition::Or { conditions }
    }
    fn csq(alias: &str, table: &str, parent: &str, child: &str) -> CorrelatedSubquery {
        CorrelatedSubquery {
            correlation: Correlation {
                parent_field: vec![parent.into()],
                child_field: vec![child.into()],
            },
            subquery: Box::new(Ast {
                table: table.into(),
                alias: Some(alias.into()),
                ..Default::default()
            }),
            system: None,
        }
    }
    fn exists(alias: &str) -> Condition {
        Condition::CorrelatedSubquery(CorrelatedSubqueryCondition {
            related: csq(alias, "track", "id", "albumId"),
            op: ExistsOp::Exists,
            flip: None,
            scalar: None,
            plan_id: None,
        })
    }
    fn albums_where(w: Condition) -> Ast {
        Ast {
            table: "album".into(),
            r#where: Some(w),
            ..Default::default()
        }
    }
    fn albums_of(artist: i64) -> Ast {
        albums_where(eq("artistId", Lit::Int(artist)))
    }

    // --- family-mate iff equal modulo holes -------------------------------------

    #[test]
    fn same_query_different_literal_shares_a_key_and_differs_in_binding() {
        let a = albums_of(1);
        let b = albums_of(2);
        assert_eq!(key(&a), key(&b));
        assert_eq!(key(&a).fingerprint(), key(&b).fingerprint());
        assert_ne!(binding(&a), binding(&b));
        assert_eq!(binding(&a), vec![CanonVal::Int(1)]);
        assert_eq!(binding(&b), vec![CanonVal::Int(2)]);
        // The exact `QueryKey`s still differ — the family is a coarser identity.
        let qa = QueryKey::new("v1", StreamMode::Normalized, &a, "vis").unwrap();
        let qb = QueryKey::new("v1", StreamMode::Normalized, &b, "vis").unwrap();
        assert_ne!(qa, qb);
    }

    #[test]
    fn identical_queries_share_key_and_binding() {
        let a = albums_of(1);
        let b = albums_of(1);
        assert_eq!(key(&a), key(&b));
        assert_eq!(binding(&a), binding(&b));
    }

    #[test]
    fn every_eligible_conjunct_is_holed_no_inference() {
        // Both equality conjuncts are holes; a conjunct whose literal agrees across the
        // family just contributes a constant column to the binding.
        let a = albums_where(and(vec![
            eq("orgId", Lit::Int(9)),
            eq("artistId", Lit::Int(1)),
        ]));
        let b = albums_where(and(vec![
            eq("orgId", Lit::Int(9)),
            eq("artistId", Lit::Int(2)),
        ]));
        let ea = extract(&a).unwrap();
        let eb = extract(&b).unwrap();
        assert_eq!(ea.key, eb.key);
        assert_eq!(
            ea.template.params,
            vec![Box::from("orgId"), Box::from("artistId")]
        );
        assert_eq!(ea.binding, vec![CanonVal::Int(9), CanonVal::Int(1)]);
        assert_eq!(eb.binding, vec![CanonVal::Int(9), CanonVal::Int(2)]);
        // Both conjuncts stripped ⇒ no residual predicate at all.
        assert_eq!(ea.template.stripped.r#where, None);
        assert_eq!(ea.template.positions, vec![0, 1]);
    }

    #[test]
    fn residual_conjuncts_stay_in_the_template() {
        // `And[a = ?, b > 3, EXISTS]` ⇒ holes [a]; residual `And[b > 3, EXISTS]`.
        let a = albums_where(and(vec![
            eq("artistId", Lit::Int(1)),
            simple("year", Op::Gt, Lit::Int(1990)),
            exists("tracks"),
        ]));
        let e = extract(&a).unwrap();
        assert_eq!(e.template.params, vec![Box::from("artistId")]);
        assert_eq!(e.template.positions, vec![0]);
        assert_eq!(
            e.template.stripped.r#where,
            Some(and(vec![
                simple("year", Op::Gt, Lit::Int(1990)),
                exists("tracks")
            ]))
        );
        // One residual conjunct collapses to itself (no singleton `And`).
        let b = albums_where(and(vec![
            eq("artistId", Lit::Int(1)),
            simple("year", Op::Gt, Lit::Int(1990)),
        ]));
        assert_eq!(
            extract(&b).unwrap().template.stripped.r#where,
            Some(simple("year", Op::Gt, Lit::Int(1990)))
        );
    }

    // --- order participates (D2) --------------------------------------------------

    #[test]
    fn conjunct_order_participates_in_the_key() {
        let ab = albums_where(and(vec![
            eq("a", Lit::Int(1)),
            simple("b", Op::Gt, Lit::Int(3)),
        ]));
        let ba = albums_where(and(vec![
            simple("b", Op::Gt, Lit::Int(3)),
            eq("a", Lit::Int(1)),
        ]));
        assert_ne!(key(&ab), key(&ba));
        // ...and so does hole order among two holes.
        let xy = albums_where(and(vec![eq("x", Lit::Int(1)), eq("y", Lit::Int(2))]));
        let yx = albums_where(and(vec![eq("y", Lit::Int(2)), eq("x", Lit::Int(1))]));
        assert_ne!(key(&xy), key(&yx));
    }

    // --- the exhaustive fallback list (design §3.2) ------------------------------

    #[test]
    fn differing_related_structure_is_a_different_family() {
        let plain = albums_of(1);
        let mut with_rel = albums_of(2);
        with_rel.related = vec![csq("tracks", "track", "id", "albumId")];
        assert_ne!(key(&plain), key(&with_rel));
    }

    #[test]
    fn differing_subquery_literal_is_a_different_family() {
        let mut a = albums_of(1);
        let mut b = albums_of(2);
        let mut ra = csq("tracks", "track", "id", "albumId");
        ra.subquery.r#where = Some(eq("genre", Lit::Int(1)));
        let mut rb = csq("tracks", "track", "id", "albumId");
        rb.subquery.r#where = Some(eq("genre", Lit::Int(2)));
        a.related = vec![ra];
        b.related = vec![rb];
        assert_ne!(key(&a), key(&b), "a subquery literal is not a hole");
        // The binding is the root hole only.
        assert_eq!(binding(&a), vec![CanonVal::Int(1)]);
    }

    #[test]
    fn order_by_limit_one_select_start_all_separate_families() {
        let base = albums_of(1);
        let mut order_by = albums_of(2);
        order_by.order_by = vec![OrderPart("title".into(), Dir::Asc)];
        let mut limit = albums_of(2);
        limit.limit = Some(10);
        let mut one = albums_of(2);
        one.one = true;
        one.limit = Some(1);
        let mut select = albums_of(2);
        select.select = Some(vec!["title".into()]);
        let mut start = albums_of(2);
        start.start = Some(Bound {
            row: BTreeMap::from([(Box::from("id"), Lit::Int(5))]),
            exclusive: true,
        });
        for (name, other) in [
            ("order_by", order_by),
            ("limit", limit),
            ("one", one),
            ("select", select),
            ("start", start),
        ] {
            assert_ne!(key(&base), key(&other), "{name} must separate families");
        }
        // A paging cursor is per-subscriber resume state, not a parameter: two
        // otherwise-identical queries with different `start` bounds are different
        // families, not two bindings.
        let mut start2 = albums_of(2);
        start2.start = Some(Bound {
            row: BTreeMap::from([(Box::from("id"), Lit::Int(6))]),
            exclusive: true,
        });
        let mut start1 = albums_of(2);
        start1.start = Some(Bound {
            row: BTreeMap::from([(Box::from("id"), Lit::Int(5))]),
            exclusive: true,
        });
        assert_ne!(key(&start1), key(&start2));
    }

    #[test]
    fn candidates_under_or_or_with_other_ops_are_not_holes() {
        // Under an `Or`: nothing is a hole ⇒ no family.
        let under_or = albums_where(or(vec![eq("a", Lit::Int(1)), eq("b", Lit::Int(2))]));
        assert!(extract(&under_or).is_none());
        // Non-`=` ops are never holes.
        for op in [
            Op::Ne,
            Op::In,
            Op::Lt,
            Op::Le,
            Op::Gt,
            Op::Ge,
            Op::Is,
            Op::IsNot,
            Op::Like,
        ] {
            let v = if op == Op::In {
                Lit::Array(vec![Lit::Int(1)])
            } else {
                Lit::Int(1)
            };
            assert!(
                extract(&albums_where(simple("a", op, v))).is_none(),
                "{op:?} must not be a hole"
            );
        }
        // …but they ride along as residual beside a real hole.
        let mixed = albums_where(and(vec![
            simple("a", Op::Ne, Lit::Int(1)),
            eq("b", Lit::Int(2)),
        ]));
        let e = extract(&mixed).unwrap();
        assert_eq!(e.template.params, vec![Box::from("b")]);
        assert_eq!(e.template.positions, vec![1]);
        assert_eq!(
            e.template.stripped.r#where,
            Some(simple("a", Op::Ne, Lit::Int(1)))
        );
        // A bare EXISTS root, or a root `Or` around a hole-shaped leaf, is a singleton.
        assert!(extract(&albums_where(exists("tracks"))).is_none());
        // A literal LHS / column RHS is not a hole.
        let lhs_lit = albums_where(Condition::Simple(SimpleCondition {
            op: Op::Eq,
            left: lit(Lit::Int(1)),
            right: lit(Lit::Int(1)),
        }));
        assert!(extract(&lhs_lit).is_none());
    }

    #[test]
    fn aggregate_group_by_having_are_refused() {
        let mut agg = albums_of(1);
        agg.aggregate = Some(rindle::Aggregate::Count);
        assert!(extract(&agg).is_none());
        let mut grouped = albums_of(1);
        grouped.group_by = vec!["artistId".into()];
        assert!(extract(&grouped).is_none());
        let mut having = albums_of(1);
        having.having = Some(eq("count", Lit::Int(1)));
        assert!(extract(&having).is_none());
    }

    #[test]
    fn null_and_array_literals_are_not_holes() {
        assert!(extract(&albums_where(eq("a", Lit::Null))).is_none());
        assert!(extract(&albums_where(eq("a", Lit::Array(vec![Lit::Int(1)])))).is_none());
        // Beside a real hole they are residual, never a binding column.
        let e = extract(&albums_where(and(vec![
            eq("a", Lit::Null),
            eq("b", Lit::Int(2)),
        ])))
        .unwrap();
        assert_eq!(e.template.params, vec![Box::from("b")]);
        assert_eq!(e.template.stripped.r#where, Some(eq("a", Lit::Null)));
    }

    #[test]
    fn no_where_is_a_singleton() {
        assert!(extract(&Ast::new("album")).is_none());
    }

    #[test]
    fn bare_exists_alias_colliding_with_related_stays_a_singleton() {
        // Concrete: `And[artistId = 1, EXISTS tracks]` + related `tracks` — the builder
        // uniquifies the gate to `tracks_0` and accepts it. Stripped: the bare EXISTS
        // keeps `tracks` and collides with the related slot ⇒ refuse here.
        let mut a = albums_where(and(vec![eq("artistId", Lit::Int(1)), exists("tracks")]));
        a.related = vec![csq("tracks", "track", "id", "albumId")];
        assert!(extract(&a).is_none());
        // A different alias does not collide.
        let mut b = albums_where(and(vec![eq("artistId", Lit::Int(1)), exists("hasTracks")]));
        b.related = vec![csq("tracks", "track", "id", "albumId")];
        assert!(extract(&b).is_some());
        // Two residual EXISTS keep the `And` (both uniquified on both sides) — no collision.
        let mut c = albums_where(and(vec![
            eq("artistId", Lit::Int(1)),
            exists("tracks"),
            exists("tracks"),
        ]));
        c.related = vec![csq("tracks", "track", "id", "albumId")];
        assert!(extract(&c).is_some());
    }

    /// The stripped template must build with the same relationship slot layout as the
    /// concrete query (the EXISTS gate's alias may lose its `_N` suffix when the `And`
    /// collapses, but slots are positional: `related` first, then gates in pre-order).
    #[test]
    fn stripped_template_keeps_the_concrete_slot_layout() {
        use rindle::{normalize_pipeline_ast, query_local_slot_names};
        let mut a = albums_where(and(vec![eq("artistId", Lit::Int(1)), exists("hasTracks")]));
        a.related = vec![csq("tracks", "track", "id", "albumId")];
        let e = extract(&a).unwrap();
        let concrete = query_local_slot_names(&normalize_pipeline_ast(&a));
        let stripped = query_local_slot_names(&normalize_pipeline_ast(&e.template.stripped));
        assert_eq!(concrete.len(), stripped.len());
        assert_eq!(concrete[0], stripped[0], "related slots are identical");
        assert_eq!(concrete[1].as_ref(), "hasTracks_0");
        assert_eq!(stripped[1].as_ref(), "hasTracks");
    }

    // --- binding canonicalization ---------------------------------------------------

    #[test]
    fn int_and_integral_number_are_one_binding() {
        let int = albums_where(eq("artistId", Lit::Int(1)));
        let num = albums_where(eq("artistId", Lit::Number(1.0)));
        let frac = albums_where(eq("artistId", Lit::Number(1.5)));
        assert_eq!(key(&int), key(&num));
        assert_eq!(binding(&int), binding(&num));
        assert_eq!(binding(&int), vec![CanonVal::Int(1)]);
        assert_eq!(key(&int), key(&frac), "the spelling is a hole either way");
        assert_ne!(binding(&int), binding(&frac));
        assert_eq!(binding(&frac), vec![CanonVal::Float(1.5f64.to_bits())]);
        // Strings and bools bind too, type-tagged.
        assert_eq!(
            binding(&albums_where(eq("name", Lit::Str("x".into())))),
            vec![CanonVal::Str("x".into())]
        );
        assert_eq!(
            binding(&albums_where(eq("flag", Lit::Bool(true)))),
            vec![CanonVal::Bool(true)]
        );
        // Different literal *types* on the same column are different bindings of the
        // same family (the key does not depend on the literal at all).
        let s = albums_where(eq("artistId", Lit::Str("1".into())));
        assert_eq!(key(&int), key(&s));
        assert_ne!(binding(&int), binding(&s));
    }

    #[test]
    fn binding_equality_agrees_with_values_equal_over_lit_pairs() {
        // The `values_equal` coherence property stated over `Lit` pairs: two literals
        // are one binding iff the scalars they lower to are join-equal.
        use rindle::value::values_equal;
        let lits = [
            Lit::Int(0),
            Lit::Int(1),
            Lit::Int(1 << 53),
            Lit::Int((1 << 53) + 1),
            Lit::Number(0.0),
            Lit::Number(1.0),
            Lit::Number(1.5),
            Lit::Number((1u64 << 53) as f64),
            Lit::Str("1".into()),
            Lit::Bool(true),
        ];
        // `values_equal` is total only within one JS type class (a number vs a string
        // is a builder bug there), so the property is stated per class.
        let same_class = |a: &Lit, b: &Lit| {
            matches!(
                (a, b),
                (Lit::Int(_) | Lit::Number(_), Lit::Int(_) | Lit::Number(_))
                    | (Lit::Str(_), Lit::Str(_))
                    | (Lit::Bool(_), Lit::Bool(_))
            )
        };
        for a in &lits {
            for b in &lits {
                let ca = canon_of_lit(a).unwrap();
                let cb = canon_of_lit(b).unwrap();
                if !same_class(a, b) {
                    assert_ne!(ca, cb, "{a:?} vs {b:?}: type-tagged keys never collide");
                    continue;
                }
                let va = ca.to_owned_value();
                let vb = cb.to_owned_value();
                assert_eq!(
                    ca == cb,
                    values_equal(va.as_ref(), vb.as_ref()),
                    "{a:?} vs {b:?}"
                );
            }
        }
    }

    // --- instantiate ∘ extract = id ---------------------------------------------------

    #[test]
    fn instantiate_is_the_exact_inverse_of_extraction() {
        let mut corpus: Vec<Ast> = vec![
            albums_of(1),
            albums_where(and(vec![
                eq("a", Lit::Int(1)),
                eq("b", Lit::Str("x".into())),
            ])),
            albums_where(and(vec![
                simple("year", Op::Gt, Lit::Int(1990)),
                eq("artistId", Lit::Int(1)),
                exists("tracks"),
                eq("genre", Lit::Bool(true)),
            ])),
            albums_where(and(vec![
                eq("a", Lit::Number(2.5)),
                eq("b", Lit::Number(3.0)),
            ])),
            table("issue")
                .r#where("assigneeId", 7i64)
                .order_by("createdAt", Dir::Desc)
                .limit(50)
                .build(),
        ];
        let mut rel = albums_where(and(vec![eq("artistId", Lit::Int(1)), exists("hasTracks")]));
        rel.related = vec![csq("tracks", "track", "id", "albumId")];
        rel.order_by = vec![OrderPart("title".into(), Dir::Asc)];
        rel.limit = Some(5);
        rel.select = Some(vec!["title".into(), "id".into()]);
        corpus.push(rel);
        for ast in &corpus {
            let e = extract(ast).unwrap();
            let mut canon = ast.clone();
            canonicalize_wire_number_lits(&mut canon);
            assert_eq!(e.template.instantiate(&e.binding), canon, "{ast:?}");
            // And a re-extraction of the instantiated form lands in the same family.
            let again = extract(&e.template.instantiate(&e.binding)).unwrap();
            assert_eq!(again.key, e.key);
            assert_eq!(again.binding, e.binding);
            assert_eq!(again.template, e.template);
        }
    }

    /// A lone residual conjunct that is itself an `And` (the generator ANDs a root
    /// equality onto an already-conjoined `where`) must round-trip with its nesting
    /// intact: the template keeps it wrapped so `instantiate` reads ONE residual, not
    /// its members. Found by `rindle-fuzz/tests/family_key_roundtrip.rs`.
    #[test]
    fn a_lone_nested_and_residual_round_trips_with_its_nesting() {
        let ast = albums_where(and(vec![
            eq("artistId", Lit::Int(2)),
            and(vec![
                exists("hasTracks"),
                simple("year", Op::IsNot, Lit::Null),
            ]),
        ]));
        let e = extract(&ast).unwrap();
        let mut canon = ast.clone();
        canonicalize_wire_number_lits(&mut canon);
        assert_eq!(e.template.instantiate(&e.binding), canon);
        assert_eq!(e.template.params.len(), 1);
        // The template's residual is the inner `And`, kept as one conjunct.
        match &e.template.stripped.r#where {
            Some(Condition::And { conditions }) => {
                assert_eq!(conditions.len(), 1);
                assert!(matches!(conditions[0], Condition::And { .. }));
            }
            other => panic!("unexpected stripped where {other:?}"),
        }
        // A hole AFTER the nested residual lands at its position too.
        let tail = albums_where(and(vec![
            and(vec![
                exists("hasTracks"),
                simple("year", Op::IsNot, Lit::Null),
            ]),
            eq("artistId", Lit::Int(2)),
        ]));
        let et = extract(&tail).unwrap();
        let mut canon_tail = tail.clone();
        canonicalize_wire_number_lits(&mut canon_tail);
        assert_eq!(et.template.instantiate(&et.binding), canon_tail);
        assert_ne!(et.key, e.key, "conjunct order is part of the key");
    }

    #[test]
    fn instantiating_another_binding_yields_that_family_mate() {
        let a = albums_where(and(vec![
            simple("year", Op::Gt, Lit::Int(1990)),
            eq("artistId", Lit::Int(1)),
        ]));
        let b = albums_where(and(vec![
            simple("year", Op::Gt, Lit::Int(1990)),
            eq("artistId", Lit::Int(2)),
        ]));
        let ea = extract(&a).unwrap();
        let eb = extract(&b).unwrap();
        assert_eq!(ea.template.instantiate(&eb.binding), b);
    }

    // --- select-order normalization still applies -------------------------------------

    #[test]
    fn select_order_does_not_fork_a_family_key() {
        let a = table("issue")
            .select("title")
            .select("priority")
            .r#where("ownerId", 1i64)
            .build();
        let b = table("issue")
            .select("priority")
            .select("title")
            .r#where("ownerId", 2i64)
            .build();
        assert_eq!(key(&a), key(&b));
    }

    #[test]
    fn schema_version_stream_mode_and_visibility_key_separate_families() {
        let a = albums_of(1);
        let base = extract_family("v1", StreamMode::Normalized, &a, "vis")
            .unwrap()
            .unwrap()
            .key;
        let sv = extract_family("v2", StreamMode::Normalized, &a, "vis")
            .unwrap()
            .unwrap()
            .key;
        let sm = extract_family("v1", StreamMode::Flat, &a, "vis")
            .unwrap()
            .unwrap()
            .key;
        let vk = extract_family("v1", StreamMode::Normalized, &a, "other")
            .unwrap()
            .unwrap()
            .key;
        assert_ne!(base, sv);
        assert_ne!(base, sm);
        assert_ne!(base, vk);
        assert_eq!(base.schema_version(), "v1");
        assert_eq!(base.visibility_key(), "vis");
        assert_eq!(base.stream_mode(), StreamMode::Normalized);
        assert_eq!(base.fingerprint_hex().len(), 16);
    }

    #[test]
    fn canonical_template_carries_positional_holes() {
        let a = albums_where(and(vec![
            simple("year", Op::Gt, Lit::Int(1990)),
            eq("artistId", Lit::Int(1)),
        ]));
        let k = key(&a);
        let text = String::from_utf8(k.canonical_template().to_vec()).unwrap();
        assert!(text.contains(r#"{"hole":0}"#), "{text}");
        assert!(
            !text.contains("\"value\":1}"),
            "the literal must not leak: {text}"
        );
        // The hole sits in the second conjunct (after `year > 1990`).
        let year = text.find("1990").unwrap();
        let hole = text.find(r#"{"hole":0}"#).unwrap();
        assert!(year < hole);
    }
}