feat(query): M2M existence tests — the two-hop correlation path

crates/ferro-schema-ir/src/lib.rs

@@ -670,6 +670,39 @@ mod tests {
         assert_eq!(encoded, ir, "query nested-exists round-trip must not drift");
     }
 
+    #[test]
+    fn query_m2m_exists_fixture_roundtrip() {
+        // The two-hop M2M existence-test golden vector (#316): the SAME
+        // exists node, a two-hop correlation path — join table first, then
+        // the target — with the scoped inner tree over the target model.
+        // Must survive a deserialize/serialize round-trip without drift.
+        let fixture =
+            include_str!("../../../tests/fixtures/ir_vectors/query_user_m2m_exists_v7.json");
+        let parsed: serde_json::Value =
+            serde_json::from_str(fixture).expect("query m2m-exists fixture must parse");
+        let ir = parsed
+            .get("ir")
+            .cloned()
+            .expect("fixture must contain ir envelope");
+        let envelope: IrEnvelope<QueryIrPayload> =
+            serde_json::from_value(ir.clone()).expect("query m2m-exists IR must deserialize");
+        match &envelope.payload.where_clause[0] {
+            QueryNode::Exists {
+                hops, where_clause, ..
+            } => {
+                assert_eq!(hops.len(), 2, "M2M correlates through two hops");
+                assert_eq!(hops[0].to_table, "tag_users");
+                assert_eq!(hops[0].to_column, "user_id");
+                assert_eq!(hops[1].from_column, "tag_id");
+                assert_eq!(hops[1].to_table, "tag");
+                assert_eq!(where_clause.len(), 1);
+            }
+            other => panic!("where[0] must be an exists node, got {other:?}"),
+        }
+        let encoded = serde_json::to_value(&envelope).expect("query m2m-exists IR must serialize");
+        assert_eq!(encoded, ir, "query m2m-exists round-trip must not drift");
+    }
+
     #[test]
     fn query_traversal_fixture_roundtrip() {
         // Multi-hop `joins` section + path-carrying leaves must survive a

src/ferro/columns.py

@@ -89,25 +89,38 @@ class RelationSpec:
 
 @dataclass(frozen=True, slots=True)
 class ReverseSpec:
-    """One reverse (BackRef) relation's existence-test facts (#314, ADR-0007).
+    """One reverse (BackRef/M2M) relation's existence-test facts (ADR-0007).
 
     Built beside :class:`RelationSpec` at the same compile choke point and
-    exposed as ``cls.__ferro_reverse_specs__``. A reverse relation supports
-    exactly one predicate form — the existence test ``t.rel.exists()`` — so
-    the spec carries only what a correlated EXISTS needs: the child model and
-    the child-side FK column the subquery correlates on. Reverse relations are
-    *tested*, never *traversed*; traversal facts stay on :class:`RelationSpec`.
+    exposed as ``cls.__ferro_reverse_specs__``. A reverse or M2M relation
+    supports exactly one predicate form — the existence test
+    ``t.rel.exists()`` — so the spec carries only what a correlated EXISTS
+    needs: the related model plus the correlation columns (the child-side FK
+    for a reverse FK; the join-table triple for M2M, #316). Reverse relations
+    are *tested*, never *traversed*; traversal facts stay on
+    :class:`RelationSpec`.
     """
 
     field_name: str
-    #: The child model — the side declaring the ``ForeignKey``.
+    #: The related model — the FK-declaring child (reverse FK) or the M2M
+    #: target: the model the inner lambda's parameter resolves against.
     target: type
-    #: Shadow FK column on the child table the EXISTS correlates against.
-    child_fk_column: str
+    #: Shadow FK column on the child table the EXISTS correlates against
+    #: (reverse FK only; ``None`` for M2M).
+    child_fk_column: str | None
     #: True when the child FK is ``unique=True`` (one-to-one BackRef). The
     #: rendering is identical at every cardinality (always correlated EXISTS);
     #: carried as a compile-side fact, not a render switch.
-    is_one_to_one: bool
+    is_one_to_one: bool = False
+    #: True for an M2M edge — the existence test correlates through the join
+    #: table (two hops) instead of a child FK (one hop).
+    is_m2m: bool = False
+    #: M2M join-table triple (``None`` for reverse FK): the association
+    #: table, its column referencing THIS side, and its column referencing
+    #: the target side — exactly as the descriptor orients them per side.
+    join_table: str | None = None
+    source_col: str | None = None
+    target_col: str | None = None
 
 
 @dataclass(frozen=True, slots=True)
@@ -495,15 +508,17 @@ def build_reverse_specs(model_cls: type[Any]) -> dict[str, ReverseSpec]:
 
     Reads the :class:`~ferro.relations.descriptors.RelationshipDescriptor`
     instances ``resolve_relationships`` installs on the class — the facts a
-    correlated EXISTS needs (child model, child FK column, one-to-one flag)
-    already live there, so this derives, never re-computes. At provisional
-    class-body time no descriptors exist yet and the map is empty; the
-    resolved second pass recompiles with descriptors installed, so the lookup
-    is complete once resolution finishes (same lifecycle as
-    :func:`build_relation_specs`).
-
-    M2M descriptors are skipped for now: the existence test's M2M form (a
-    two-hop correlation through the join table) lands in #316.
+    correlated EXISTS needs (related model, child FK column or join-table
+    triple, one-to-one flag) already live there, so this derives, never
+    re-computes. At provisional class-body time no descriptors exist yet and
+    the map is empty; the resolved second pass recompiles with descriptors
+    installed, so the lookup is complete once resolution finishes (same
+    lifecycle as :func:`build_relation_specs`).
+
+    M2M descriptors (#316) map to a join-table spec; each side's descriptor
+    already orients ``source_col``/``target_col`` for that side, so the spec
+    is a straight copy of resolution's facts on both the declaring and the
+    ``related_name`` side.
 
     Callers store the result on ``cls.__ferro_reverse_specs__`` and must
     replace, never mutate — same convention as ``__ferro_columns__``.
@@ -518,11 +533,20 @@ def build_reverse_specs(model_cls: type[Any]) -> dict[str, ReverseSpec]:
         for name, attr in vars(klass).items():
             if not isinstance(attr, RelationshipDescriptor) or name in specs:
                 continue
-            if attr.is_m2m:
-                continue
             target = REGISTRY.resolve_reference(attr.target_model_name, default=None)
             if target is None:
                 continue
+            if attr.is_m2m:
+                specs[name] = ReverseSpec(
+                    field_name=name,
+                    target=target,
+                    child_fk_column=None,
+                    is_m2m=True,
+                    join_table=attr.join_table,
+                    source_col=attr.source_col,
+                    target_col=attr.target_col,
+                )
+                continue
             specs[name] = ReverseSpec(
                 field_name=name,
                 target=target,

src/ferro/query/nodes.py

@@ -1010,15 +1010,15 @@ def _collect_inner_traversal_paths(
 
 
 class ReverseRelationProxy:
-    """Predicate proxy for a reverse (BackRef) relation (#314, ADR-0007).
+    """Predicate proxy for a reverse (BackRef) or M2M relation (ADR-0007).
 
     Returned by attribute access on a :class:`QueryProxy` when the accessed
-    name is a resolved reverse relation. It exposes exactly one verb — the
-    existence test :meth:`exists` — because reverse relations are *tested*,
-    never *traversed*: column access, comparisons (including ``!= None`` /
-    ``== None``), and ``in_`` raise at build time with the supported spelling
-    in the message. Negation is uniform ``~`` over the returned node
-    (ADR-0008), so NOT EXISTS is ``~t.rel.exists()``.
+    name is a resolved reverse or many-to-many relation. It exposes exactly
+    one verb — the existence test :meth:`exists` — because reverse relations
+    are *tested*, never *traversed*: column access, comparisons (including
+    ``!= None`` / ``== None``), and ``in_`` raise at build time with the
+    supported spelling in the message. Negation is uniform ``~`` over the
+    returned node (ADR-0008), so NOT EXISTS is ``~t.rel.exists()``.
     """
 
     __slots__ = ("_root_model", "_name", "_spec")
@@ -1069,13 +1069,46 @@ def exists(self, inner: "Predicate[Any] | None" = None) -> QueryNode:
                 "column: the existence test correlates the child's FK "
                 "against the root primary key."
             )
-        hop = QueryJoinHop(
-            relation=self._name,
-            from_column=root_pk,
-            to_table=self._spec.target.__ferro_table__,
-            to_column=self._spec.child_fk_column,
-            target=self._spec.target,
-        )
+        if self._spec.is_m2m:
+            # Two hops through the join table (#316), same node and render
+            # loop as the one-hop form: the join table correlates to the
+            # enclosing scope, the target joins on inside the subquery. Both
+            # hops carry this relation's name — one relation, one alias
+            # family.
+            target_pk = getattr(self._spec.target, "__ferro_pk__", None)
+            if target_pk is None:
+                raise ValueError(
+                    f"t.{self._name}.exists() requires "
+                    f"{self._spec.target.__name__} to declare a primary-key "
+                    "column: the M2M existence test joins the join table to "
+                    "the target primary key."
+                )
+            hops = (
+                QueryJoinHop(
+                    relation=self._name,
+                    from_column=root_pk,
+                    to_table=self._spec.join_table,
+                    to_column=self._spec.source_col,
+                    target=self._spec.target,
+                ),
+                QueryJoinHop(
+                    relation=self._name,
+                    from_column=self._spec.target_col,
+                    to_table=self._spec.target.__ferro_table__,
+                    to_column=target_pk,
+                    target=self._spec.target,
+                ),
+            )
+        else:
+            hops = (
+                QueryJoinHop(
+                    relation=self._name,
+                    from_column=root_pk,
+                    to_table=self._spec.target.__ferro_table__,
+                    to_column=self._spec.child_fk_column,
+                    target=self._spec.target,
+                ),
+            )
         where: tuple[QueryNode, ...] = ()
         joins: tuple[Any, ...] = ()
         if inner is not None:
@@ -1093,7 +1126,7 @@ def exists(self, inner: "Predicate[Any] | None" = None) -> QueryNode:
             where = (node,)
         return QueryNode(
             exists=ExistsTest(
-                hops=(hop,), where=where, joins=joins, owner=self._root_model
+                hops=hops, where=where, joins=joins, owner=self._root_model
             )
         )
 

src/query.rs

@@ -1268,6 +1268,59 @@ mod tests {
         );
     }
 
+    #[test]
+    fn two_hop_exists_renders_join_table_then_target() {
+        // M2M existence test (#316): the SAME render loop — first hop's
+        // table is the subquery FROM correlated to the enclosing scope,
+        // the second hop renders as an inner join inside the subquery, and
+        // the inner tree qualifies by the LAST hop's alias (the target).
+        let payload: ferro_schema_ir::QueryIrPayload = serde_json::from_value(json!({
+            "model_name": "User",
+            "where": [
+                {"node_kind": "exists",
+                 "hops": [
+                    {"relation": "tags", "from_column": "id",
+                     "to_table": "tag_users", "to_column": "user_id"},
+                    {"relation": "tags", "from_column": "tag_id",
+                     "to_table": "tag", "to_column": "id"}
+                 ],
+                 "where": [
+                    {"node_kind": "leaf", "column": "name", "operator": "==",
+                     "value": {"kind": "string", "value": "admin"}, "path": []}
+                 ]}
+            ],
+            "order_by": [],
+            "limit": null, "offset": null, "m2m": null, "materialization": {"kind": "root_instances"}, "joins": []
+        }))
+        .expect("payload deserializes");
+        let plan = QueryPlan::from_ir_payload(payload).expect("plan builds");
+        let mut select = Query::select();
+        select.from(Alias::new("user")).cond_where(
+            plan.to_condition_for_backend(Dialect::Postgres, Some("user"))
+                .expect("valid test query"),
+        );
+        let sql = select.to_string(PostgresQueryBuilder).to_lowercase();
+        assert!(
+            sql.contains("exists(select 1 from \"tag_users\" as \"x1_tags\""),
+            "join table is the subquery FROM: {sql}"
+        );
+        assert!(
+            sql.contains(
+                "inner join \"tag\" as \"x2_tags\" on \
+                 \"x1_tags\".\"tag_id\" = \"x2_tags\".\"id\""
+            ),
+            "target joins on inside the subquery: {sql}"
+        );
+        assert!(
+            sql.contains("\"x1_tags\".\"user_id\" = \"user\".\"id\""),
+            "join table correlates to the enclosing alias: {sql}"
+        );
+        assert!(
+            sql.contains("\"x2_tags\".\"name\" = 'admin'"),
+            "inner tree qualifies by the LAST hop's alias: {sql}"
+        );
+    }
+
     #[test]
     fn exists_requires_a_qualified_enclosing_scope() {
         // Correlation needs an alias to correlate against; the unqualified

tests/fixtures/ir_vectors/query_user_m2m_exists_v7.json

@@ -0,0 +1,51 @@
+{
+  "vector_name": "query_user_m2m_exists_v7",
+  "domain": "query",
+  "expect_valid": true,
+  "ir": {
+    "ir_kind": "query",
+    "ir_version": 7,
+    "payload": {
+      "model_name": "User",
+      "where": [
+        {
+          "node_kind": "exists",
+          "hops": [
+            {
+              "relation": "tags",
+              "from_column": "id",
+              "to_table": "tag_users",
+              "to_column": "user_id"
+            },
+            {
+              "relation": "tags",
+              "from_column": "tag_id",
+              "to_table": "tag",
+              "to_column": "id"
+            }
+          ],
+          "where": [
+            {
+              "node_kind": "leaf",
+              "column": "name",
+              "operator": "==",
+              "value": {
+                "kind": "string",
+                "value": "admin"
+              },
+              "path": []
+            }
+          ]
+        }
+      ],
+      "order_by": [],
+      "limit": null,
+      "offset": null,
+      "m2m": null,
+      "joins": [],
+      "materialization": {
+        "kind": "root_instances"
+      }
+    }
+  }
+}