From 1347e22be9e376869bf1dd1cb323ac7306990e82 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 2 Oct 2026 10:20:39 +0000
Subject: [PATCH 1/4] feat(query-builder): resolve joined tables to their alias
 when the query runs

A builder that is returned and extended later can't use condition_and
anymore, and a plain where() on a joined table used to raise
AliasedTableMismatchError. Fields of a table that is joined under exactly
one alias are now pointed at that alias in where/orderby/groupby/having,
at collect time, so where().join() and join().where() behave the same.

- A table joined more than once still raises; pick the join with a
  lambda whose extra arguments name the relationship
  (`where(lambda article, reviewer: ...)`, nested as `parent__child`).
  These lambdas are resolved at collect time too.
- The limitby id subquery and count() include the left joins the
  predicate filters on, so pagination and counts match the rows.
- The distinct-ids pagination subquery names its id column, which no
  longer collides with an ordered-by id of a joined table.
- delete()/update() refuse relationship lambdas instead of dropping them.
- Cache keys use relationship paths instead of per-process alias hashes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FXZN6UgVPuRxvHzMMTShAE
---
 docs/3_building_queries.md   |  29 ++-
 src/typedal/query_builder.py | 374 ++++++++++++++++++++++++++++++-----
 tests/test_join_aliases.py   | 256 ++++++++++++++++++++++++
 tests/test_query_builder.py  |  21 +-
 4 files changed, 613 insertions(+), 67 deletions(-)
 create mode 100644 tests/test_join_aliases.py

diff --git a/docs/3_building_queries.md b/docs/3_building_queries.md
index 9c94155..098a58a 100644
--- a/docs/3_building_queries.md
+++ b/docs/3_building_queries.md
@@ -195,19 +195,34 @@ Person.cross_join(Color)  # every person combined with every color
 Person.cross_join(Article).where(Article.id == Comment.article)  # comment is linked via article
 ```
 
-`AliasedTableMismatchError` (a subclass of `ImplicitCrossJoinError`) is raised when you filter on a table
-that is also joined through a relationship. A relationship join uses an alias, so `where(Article.x > 0)` refers to a
-*second*, unjoined copy of `article`. Put the condition on the relationship instead:
+A relationship join puts the related table in the SQL under an alias. Filtering or ordering on that table
+(`where()`, `orderby()`, `groupby()`, `having()`) is pointed at the alias when the query runs, so it doesn't matter
+whether `where()` comes before or after `join()`, or whether the builder is returned and extended somewhere else:
 
 ```python
-# wrong: Article here is not the joined (aliased) articles table
 Person.join("articles", method="inner").where(Article.published == True)
+Person.where(Article.published == True).join("articles", method="inner")  # the same query
+```
+
+Like `condition_and`, this filters the joined rows too: each person comes back with only their published articles.
+On a left join it also drops the persons without a matching article, and `where(Article.id == None)` finds the persons
+without any. Pagination and `count()` take the filter into account.
+
+When the same table is joined more than once, `Article` alone is ambiguous and raises `AliasedTableMismatchError`
+(a subclass of `ImplicitCrossJoinError`). Pick the join with extra lambda arguments, named after the relationship
+(a nested one as `parent__child`, or just `child` when that name is unique). These are resolved when the query runs,
+so they can also come before the `join()`:
 
-# right:
-Person.join("articles", method="inner", condition_and=lambda person, article: article.published == True)
+```python
+builder = Article.join("writer").join("reviewer")  # both relationships to Author
+builder.where(Author.name == "ann")  # raises: author is joined as 'writer' and as 'reviewer'
+builder.where(lambda article, reviewer: reviewer.name == "ann")
 ```
 
-If you do want that second, independent copy, `cross_join(Article)` makes it explicit and exempts it from this check.
+`delete()` and `update()` ignore joins, so they refuse a builder with such a lambda.
+
+If you want a second, independent copy of a joined table, `cross_join(Article)` makes it explicit: the table name
+then means that copy, not the joined one.
 
 ### groupby & having
 
diff --git a/src/typedal/query_builder.py b/src/typedal/query_builder.py
index c7296a3..eed4a73 100644
--- a/src/typedal/query_builder.py
+++ b/src/typedal/query_builder.py
@@ -4,8 +4,10 @@
 
 from __future__ import annotations
 
+import copy
 import datetime as dt
 import functools
+import inspect
 import math
 import time
 import typing as t
@@ -100,6 +102,63 @@ def _walk_tables(node: object, tables: set[str], parents: dict[str, str]) -> set
     return found
 
 
+class _JoinedTable(t.NamedTuple):
+    """A relationship table as it ends up in the SQL of a builder with joins."""
+
+    path: str  # e.g. "users" or "users.bestie"
+    relation: Relationship[t.Any]
+    table: Table  # aliased when the relationship is joined under an alias
+    tablename: str  # the original, unaliased table name
+    aliased: bool
+
+
+def _rewrite_tables(node: t.Any, tables: dict[str, Table]) -> t.Any:
+    """
+    Rebuild node (a query, expression or field) with every field of a table in `tables` pointing at its alias.
+
+    Unchanged branches are returned as-is, so a query without such fields comes back as the same object.
+    """
+    if isinstance(node, Field):
+        alias = tables.get(t.cast(str, node.tablename))
+        return node if alias is None else alias[node.name]
+    if isinstance(node, (list, tuple)):
+        items = [_rewrite_tables(item, tables) for item in node]
+        changed = any(new is not old for new, old in zip(items, node, strict=True))
+        return type(node)(items) if changed else node
+    if not isinstance(node, (Expression, Query)):
+        return node
+
+    first = _rewrite_tables(node.first, tables)
+    second = _rewrite_tables(node.second, tables)
+    if first is node.first and second is node.second:
+        return node
+
+    clone = copy.copy(node)
+    clone.first = first
+    clone.second = second
+    if isinstance(clone, Expression):
+        # as Expression.__init__ derives it:
+        clone._table = getattr(first, "_table", None)
+    return clone
+
+
+def _requested_tables(part: t.Any) -> list[str]:
+    """
+    The joined tables a where() lambda asks for: its arguments after the model, e.g. `lambda article, tags: ...`.
+
+    Arguments with a default don't ask for anything.
+    """
+    if not callable(part) or isinstance(part, (Field, Query, Expression, dict)) or is_typed_field(part):
+        return []
+    try:
+        parameters = inspect.signature(part).parameters.values()
+    except (TypeError, ValueError):  # pragma: no cover - builtins without a signature
+        return []
+    positional = (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD)
+    names = [p.name for p in parameters if p.kind in positional and p.default is inspect.Parameter.empty]
+    return names[1:]
+
+
 def warn_noop(method: str) -> None:
     """
     Warn that `method` was called without arguments, which has no effect.
@@ -125,6 +184,8 @@ class QueryBuilder[T_MetaInstance: _TypedTable](Select):
     metadata: Metadata
     _permissions: Permissions
     cross_joins: list[t.Type[TypedTable]]
+    # where() calls with a lambda asking for joined tables; resolved at collect time, ANDed with `query`:
+    deferred_queries: list[tuple[t.Any, ...]]
 
     def __setattr__(self, key: str, value: t.Any) -> None:
         """Keep QueryBuilder state independent from Select's table-like storage."""
@@ -155,6 +216,7 @@ def __init__(
         metadata: Metadata | None = None,
         permissions: Permissions | None = None,
         cross_joins: list[t.Type[TypedTable]] | None = None,
+        deferred_queries: list[tuple[t.Any, ...]] | None = None,
     ):
         """
         Normally, you wouldn't manually initialize a QueryBuilder but start using a method on a TypedTable.
@@ -172,6 +234,7 @@ def __init__(
         self.metadata = metadata or {}
         self._permissions = merge_permissions(getattr(model, "_permissions", None), permissions)
         self.cross_joins = cross_joins or []
+        self.deferred_queries = deferred_queries or []
 
     def _ensure_table_defined(self) -> Table:
         model = self.model
@@ -215,6 +278,7 @@ def __bool__(self) -> bool:
                 self.relationships,
                 self.metadata,
                 self.cross_joins,
+                self.deferred_queries,
             ],
         )
 
@@ -228,6 +292,7 @@ def _extend(
         metadata: Metadata | None = None,
         permissions: Permissions | None = None,
         cross_joins: list[t.Type[TypedTable]] | None = None,
+        deferred_queries: list[tuple[t.Any, ...]] | None = None,
     ) -> "QueryBuilder[T_MetaInstance]":
         return QueryBuilder(
             self.model,
@@ -238,6 +303,7 @@ def _extend(
             (self.metadata | (metadata or {})) if metadata else self.metadata,  # ty: ignore[invalid-argument-type]
             permissions=merge_permissions(self._permissions, permissions),
             cross_joins=self.cross_joins + (cross_joins or []),
+            deferred_queries=self.deferred_queries + (deferred_queries or []),
         )
 
     def permissions(self, **permissions: t.Unpack[Permissions]) -> "QueryBuilder[T_MetaInstance]":
@@ -346,7 +412,7 @@ def having(self, condition: t.Any) -> "QueryBuilder[T_MetaInstance]":
 
     def where(
         self,
-        *queries_or_lambdas: Query | t.Callable[[t.Type[T_MetaInstance]], Query] | dict[str, t.Any],
+        *queries_or_lambdas: Query | t.Callable[..., Query] | dict[str, t.Any],
         **filters: t.Any,
     ) -> "QueryBuilder[T_MetaInstance]":
         """
@@ -362,6 +428,12 @@ def where(
         When passing multiple queries to a single .where, they will be ORed:
             .where(lambda table: table.id == 5, lambda table: table.id == 6) == (table.id == 5) | (table.id=6)
 
+        Joined relationships live under an alias in the SQL. A field of a joined table (`.where(Tag.name == "x")`)
+        is resolved to that alias when the builder runs, so `.where(...).join(...)` and `.join(...).where(...)` both
+        work. When the same table is joined more than once, pick the join with extra lambda arguments, named after
+        the relationship (nested ones as `parent__child`), which are also resolved when the builder runs:
+            .join("tags").where(lambda article, tags: tags.name == "x")
+
         Calling this without any arguments on a builder that already has settings
         does nothing and emits a NoopQueryWarning.
         Starting an empty builder (e.g. `Model.where()`) is allowed and stays silent.
@@ -370,14 +442,28 @@ def where(
             warn_noop("where")
             return self
 
-        new_query = self.query
-        table = self._ensure_table_defined()
-
         queries_or_lambdas = (
             *queries_or_lambdas,
             filters,
         )
 
+        if any(_requested_tables(query_part) for query_part in queries_or_lambdas):
+            # the joined tables (and their aliases) are only known once the builder runs:
+            return self._extend(deferred_queries=[queries_or_lambdas])
+
+        new_query = self.query
+        if subquery := self._where_group(queries_or_lambdas, {}):
+            new_query &= subquery
+
+        return self._extend(overwrite_query=new_query)
+
+    def _where_group(self, queries_or_lambdas: tuple[t.Any, ...], joined: dict[str, Table]) -> Query:
+        """
+        OR the parts of one where() call together.
+
+        `joined` maps relationship names to their (aliased) tables, for lambdas that ask for them by name.
+        """
+        table = self._ensure_table_defined()
         subquery = t.cast(Query, DummyQuery())
         for query_part in queries_or_lambdas:
             if isinstance(query_part, Field) or is_typed_field(query_part):
@@ -385,7 +471,7 @@ def where(
             elif isinstance(query_part, (Query, Expression)):
                 subquery |= t.cast(Query, query_part)
             elif callable(query_part):
-                if result := query_part(self.model):
+                if result := query_part(self.model, *self._bind_joined_tables(query_part, joined)):
                     subquery |= t.cast(Query, result)
             elif isinstance(query_part, dict):
                 subsubquery = DummyQuery()
@@ -397,10 +483,15 @@ def where(
             else:
                 raise ValueError(f"Unexpected query type ({type(query_part)}).")
 
-        if subquery:
-            new_query &= subquery
+        return subquery
 
-        return self._extend(overwrite_query=new_query)
+    def _bind_joined_tables(self, fn: t.Callable[..., t.Any], joined: dict[str, Table]) -> list[Table]:
+        """Look up the joined tables that a where() lambda asks for, by the names of its extra arguments."""
+        names = _requested_tables(fn)
+        if missing := [name for name in names if name not in joined]:
+            available = ", ".join(sorted(joined)) or "none"
+            raise ValueError(f"where() asks for unjoined relationship(s) {', '.join(missing)} (joined: {available})")
+        return [joined[name] for name in names]
 
     def _parse_relationships(
         self,
@@ -611,14 +702,25 @@ def _select_arg_convert(self, arg: t.Any) -> t.Any:
 
         return arg
 
+    def _mutation_query(self) -> Query:
+        """
+        The query for delete and update, which ignore joins.
+
+        A where() lambda that asks for joined tables only filters through those joins, so leaving it out would
+        delete or update more rows than asked for.
+        """
+        if self.deferred_queries:
+            raise ValueError("delete() and update() can't use where() lambdas that ask for joined relationships")
+        return self.query
+
     def delete(self) -> list[int]:
         """
         Based on the current query, delete rows and return a list of deleted IDs.
         """
         require_permission(self._permissions, "delete")
         db = self._get_db()
-        removed_ids = [_.id for _ in db(self.query).select("id")]
-        if db(self.query).delete():
+        removed_ids = [_.id for _ in db(self._mutation_query()).select("id")]
+        if db(self._mutation_query()).delete():
             # success!
             return removed_ids
 
@@ -626,7 +728,7 @@ def delete(self) -> list[int]:
 
     def _delete(self) -> str:
         db = self._get_db()
-        return str(db(self.query)._delete())
+        return str(db(self._mutation_query())._delete())
 
     # QueryBuilder subclasses Select for its query-building surface but yields
     # model instances rather than fields, so these two are not substitutable.
@@ -640,16 +742,115 @@ def update(self, **fields: t.Any) -> list[int]:  # ty: ignore[invalid-method-ove
         from .updates import UpdateSet
 
         # a TypedTable's primary key is always its integer id
-        return t.cast(list[int], t.cast(UpdateSet, db(self.query)).update_ids(**fields))
+        return t.cast(list[int], t.cast(UpdateSet, db(self._mutation_query())).update_ids(**fields))
 
     def _update(self, **fields: t.Any) -> str:
         db = self._get_db()
-        return str(db(self.query)._update(**fields))
+        return str(db(self._mutation_query())._update(**fields))
+
+    def _joined_tables(self) -> list[_JoinedTable]:
+        """Every joined relationship (nested ones included) with the table it uses in the SQL."""
+        joined: list[_JoinedTable] = []
+        for key, relation in self.relationships.items():
+            self._collect_joined_tables(relation, key, self.model, key, joined)
+        return joined
+
+    def _collect_joined_tables(
+        self,
+        relation: Relationship[t.Any],
+        key: str,
+        parent: t.Any,
+        path: str,
+        joined: list[_JoinedTable],
+    ) -> None:
+        # mirrors the aliasing in _process_relationship_for_left_join and _build_inner_joins_recursive:
+        # everything is aliased, except an `on` relationship to another table.
+        other = relation.get_table(self._get_db())
+        aliased = other is parent or not relation.on
+        table = other.with_alias(f"{key}_{hash(relation)}") if aliased else other
+        joined.append(_JoinedTable(path, relation, t.cast(Table, table), other._tablename, aliased))
+
+        for nested_name, nested in relation.nested.items():
+            self._collect_joined_tables(nested, nested_name, table, f"{path}.{nested_name}", joined)
+
+    def _alias_rewrites(self, joined: list[_JoinedTable]) -> dict[str, Table]:
+        """
+        Map table names to the one alias they are joined under.
+
+        Left out (so the table name keeps meaning the table itself): the root table, `cross_join()` tables,
+        tables joined without an alias (`on=`) and tables joined more than once, which _validate_joins rejects.
+        """
+        excluded = {self._ensure_table_defined()._tablename}
+        excluded |= {table._ensure_table_defined()._tablename for table in self.cross_joins}
+
+        per_table: dict[str, list[_JoinedTable]] = defaultdict(list)
+        for entry in joined:
+            per_table[entry.tablename].append(entry)
+
+        return {
+            name: entries[0].table
+            for name, entries in per_table.items()
+            if len(entries) == 1 and entries[0].aliased and name not in excluded
+        }
+
+    @staticmethod
+    def _joined_by_name(joined: list[_JoinedTable]) -> dict[str, Table]:
+        """The names a where() lambda can ask for: `tags`, `tags__author` and, when unambiguous, `author`."""
+        names = {entry.path.replace(".", "__"): entry.table for entry in joined}
+
+        per_leaf: dict[str, list[Table]] = defaultdict(list)
+        for entry in joined:
+            per_leaf[entry.path.rsplit(".", 1)[-1]].append(entry.table)
+        for name, tables in per_leaf.items():
+            if len(tables) == 1:
+                names.setdefault(name, tables[0])
+
+        return names
+
+    def _resolve_query(self, joined: list[_JoinedTable], rewrites: dict[str, Table]) -> Query:
+        """The builder's query with the deferred where() lambdas applied and joined tables pointing at their alias."""
+        query = self.query
+        if self.deferred_queries:
+            by_name = self._joined_by_name(joined)
+            for queries_or_lambdas in self.deferred_queries:
+                if subquery := self._where_group(queries_or_lambdas, by_name):
+                    query &= subquery
+
+        return t.cast(Query, _rewrite_tables(query, rewrites)) if rewrites else query
+
+    @staticmethod
+    def _required_join_names(joined: list[_JoinedTable], *nodes: t.Any) -> set[str]:
+        """
+        The (aliased) names of the joined tables that `nodes` (a predicate, a counted field) use.
+
+        Includes the joins they hang from, since a nested join's ON clause references its parent's join.
+        """
+        referenced: set[str] = set()
+        for node in nodes:
+            referenced |= _walk_tables(node, set(), {})
+        paths = {entry.path for entry in joined if entry.table._tablename in referenced}
+        paths |= {path.rsplit(".", depth)[0] for path in paths for depth in range(1, path.count(".") + 1)}
+        return {entry.table._tablename for entry in joined if entry.path in paths}
+
+    @staticmethod
+    def _required_left_joins(names: set[str], left_joins: list[Expression]) -> list[Expression]:
+        """
+        The left joins of the tables in `names`, see _required_join_names.
+
+        Queries that replace the select (the id subquery of limitby, count) need these: without them, a filter on a
+        left-joined table references a table that isn't in their FROM clause.
+        """
+        return [join for join in left_joins if isinstance(join.first, Table) and join.first._tablename in names]
 
     def _before_query(self, mut_metadata: Metadata, add_id: bool = True) -> tuple[Query, list[t.Any], SelectKwargs]:
         select_args = [self._select_arg_convert(_) for _ in self.select_args] or [self.model.ALL]
         select_kwargs = self.select_kwargs.copy()
-        query = self.query
+        joined = self._joined_tables()
+        rewrites = self._alias_rewrites(joined)
+        query = self._resolve_query(joined, rewrites)
+        for option in ("orderby", "groupby", "having"):
+            if rewrites and option in select_kwargs:
+                select_kwargs[option] = _rewrite_tables(select_kwargs[option], rewrites)
         model = self.model
         mut_metadata["query"] = query
         # require at least id of main table:
@@ -661,9 +862,11 @@ def _before_query(self, mut_metadata: Metadata, add_id: bool = True) -> tuple[Qu
             select_args.append(model.id)
 
         if self.relationships:
-            query, select_args = self._handle_relationships_pre_select(query, select_args, select_kwargs, mut_metadata)
+            query, select_args = self._handle_relationships_pre_select(
+                query, select_args, select_kwargs, mut_metadata, joined
+            )
         else:
-            self._validate_joins([], select_args, select_kwargs)
+            self._validate_joins(query, [], select_args, select_kwargs, joined)
 
         for table in self.cross_joins:
             query &= table.id > 0
@@ -672,14 +875,16 @@ def _before_query(self, mut_metadata: Metadata, add_id: bool = True) -> tuple[Qu
 
     def _validate_joins(
         self,
+        query: Query,
         conditions: list[QueryLike],
         fields: list[t.Any],
         options: SelectKwargs,
+        joined: list[_JoinedTable],
     ) -> None:
         """
         Reject tables that would end up in the FROM clause without anything relating them to the root table.
 
-        Walks the user's predicate, the extra `conditions`, the selected `fields` and the ON clauses once.
+        Walks the (resolved) predicate, the extra `conditions`, the selected `fields` and the ON clauses once.
         Every table referenced inside a single comparison (`A.x == B.y`, but also `(A.x + B.y) > 3`) counts as
         linked. The root table, explicit JOIN targets and `cross_join()` tables seed the connected component.
         """
@@ -691,7 +896,7 @@ def _validate_joins(
         parents: dict[str, str] = {}
         walk = functools.partial(_walk_tables, tables=tables, parents=parents)
 
-        predicate_tables = walk(self.query)
+        predicate_tables = walk(query)
         for condition in conditions:
             walk(condition)
         for field in fields:
@@ -706,13 +911,20 @@ def _validate_joins(
                 seeds.add(join.first._tablename)
             walk(join.second)
 
-        db = self._get_db()
-        for name, relation in self.relationships.items():
-            if not relation.condition:
+        # _alias_rewrites resolved every table joined under exactly one alias, what's left is ambiguous:
+        per_table: dict[str, list[_JoinedTable]] = defaultdict(list)
+        for entry in joined:
+            per_table[entry.tablename].append(entry)
+        for original, entries in per_table.items():
+            if original not in predicate_tables or original == root or original in explicit:
                 continue
-            original = relation.get_table(db)._tablename
-            if original in predicate_tables and original != root and original not in explicit:
-                raise AliasedTableMismatchError(f"Table {original!r} is joined under alias {name!r}")
+            if all(entry.aliased for entry in entries):
+                paths = ", ".join(repr(entry.path) for entry in entries)
+                pick = entries[0].path.replace(".", "__")
+                raise AliasedTableMismatchError(
+                    f"Table {original!r} is joined under multiple aliases ({paths}); "
+                    f"pick one with .where(lambda row, {pick}: ...)"
+                )
 
         tables |= seeds
         _link_tables(parents, seeds)
@@ -755,7 +967,7 @@ def _collect_cached(
             self.model,
             f"{into.__module__}.{into.__qualname__}",
             metadata,
-            self.query,
+            self._cache_key_query(),
             self.select_args,
             self.select_kwargs,
             self.relationships.keys(),
@@ -767,6 +979,19 @@ def _collect_cached(
 
         return load_from_cache(key, self._get_db())
 
+    def _cache_key_query(self) -> Query | str:
+        """The resolved query, with alias names (which contain a per-process hash) replaced by relationship paths."""
+        joined = self._joined_tables()
+        query = self._resolve_query(joined, self._alias_rewrites(joined))
+        if query is self.query:
+            return query
+
+        key = str(query)
+        for entry in joined:
+            if entry.aliased:
+                key = key.replace(entry.table._tablename, f"<{entry.path}>")
+        return key
+
     def execute(self, add_id: bool = False) -> Rows:
         """
         Raw version of .collect which only executes the SQL, without performing t.Any magic afterwards.
@@ -930,6 +1155,7 @@ def _handle_relationships_pre_select(
         select_args: list[t.Any],
         select_kwargs: SelectKwargs,
         metadata: Metadata,
+        joined: list[_JoinedTable],
     ) -> tuple[Query, list[t.Any]]:
         """Handle relationship joins and field selection for database query."""
         # Collect all relationship keys including nested ones
@@ -941,8 +1167,15 @@ def _handle_relationships_pre_select(
         select_args = self._build_left_joins_and_fields(select_args, left_joins)
 
         # validate before the limitby optimization replaces the predicate with an ID subquery:
-        self._validate_joins([], select_args, {"join": inner_joins or select_kwargs.get("join"), "left": left_joins})
-        query = self._apply_limitby_optimization(query, select_kwargs, inner_joins, metadata)
+        self._validate_joins(
+            query,
+            [],
+            select_args,
+            {"join": inner_joins or select_kwargs.get("join"), "left": left_joins},
+            joined,
+        )
+        required_left = self._required_left_joins(self._required_join_names(joined, query), left_joins)
+        query = self._apply_limitby_optimization(query, select_kwargs, inner_joins, metadata, required_left)
 
         if inner_joins:
             select_kwargs["join"] = inner_joins
@@ -1046,7 +1279,8 @@ def _select_distinct_ids_with_orderby_fields(self, query: Query, select_kwargs:
         model = self.model
         id_field = t.cast(TypedField[int], model.id)
 
-        select_args: list[OrderBy] = [id_field]
+        # named, because a joined table's id can be among the orderby fields too:
+        select_args: list[OrderBy] = [t.cast(OrderBy, id_field.with_alias("typedal_paginate_id"))]
         seen = {str(model.id)}
 
         for field in self._selectable_orderby_fields(select_kwargs.get("orderby")):
@@ -1056,9 +1290,8 @@ def _select_distinct_ids_with_orderby_fields(self, query: Query, select_kwargs:
                 seen.add(key)
 
         ids = db(query)._select(*select_args, **select_kwargs).rstrip(";")
-        id_column = getattr(model.id, "_raw_rname", id_field.name)
-        return f'SELECT "{id_column}" FROM ({ids}) AS typedal_paginate_ids'  # noqa: S608
-        # id_column originates from code
+        id_column = db._adapter.dialect.quote("typedal_paginate_id")
+        return f"SELECT {id_column} FROM ({ids}) AS typedal_paginate_ids"  # noqa: S608
         # ids is a safe subquery, originating from code
 
     def _apply_limitby_optimization(
@@ -1067,8 +1300,13 @@ def _apply_limitby_optimization(
         select_kwargs: SelectKwargs,
         joins: list[t.Any],
         metadata: Metadata,
+        left_joins: list[Expression] | None = None,
     ) -> Query:
-        """Apply limitby optimization when relationships are present."""
+        """
+        Apply limitby optimization when relationships are present.
+
+        `left_joins` are the left joins the predicate filters on, see _required_left_joins.
+        """
         if not (limitby := select_kwargs.pop("limitby", ())):
             return query
 
@@ -1080,9 +1318,15 @@ def _apply_limitby_optimization(
 
         if joins:
             kwargs["join"] = joins
+        if left_joins:
+            kwargs["left"] = left_joins
+        if joins or left_joins:
             kwargs["distinct"] = True
+        if left_joins and not kwargs.get("orderby") and kwargs.get("orderby_on_limitby", True):
+            # pydal's implicit limitby order includes the left-joined ids, which DISTINCT doesn't select:
+            kwargs["orderby"] = t.cast(OrderBy, model.id)
 
-        if joins and kwargs.get("orderby"):
+        if (joins or left_joins) and kwargs.get("orderby"):
             ids = self._select_distinct_ids_with_orderby_fields(query, kwargs)
         else:
             ids = db(query)._select(model.id, **kwargs)
@@ -1431,10 +1675,39 @@ def __count(
         distinct: bool | Field | TypedField[t.Any] = False,
         *,
         include_left_for_distinct: bool = True,
-    ) -> Query:
-        # internal, shared logic between .count and ._count
+    ) -> tuple[Query | str, bool | Field | TypedField[t.Any]]:
+        """
+        Internal, shared logic between .count and ._count.
+
+        Returns the query to count, or the full SQL when the query filters on a left-joined table
+        (pydal's count can't LEFT JOIN), plus the `distinct` to count with.
+        """
         model = self.model
-        query = self.query
+        joined = self._joined_tables()
+        rewrites = self._alias_rewrites(joined)
+        query = self._resolve_query(joined, rewrites)
+        if rewrites and isinstance(distinct, (Field, Expression)):
+            distinct = _rewrite_tables(distinct, rewrites)
+
+        for table in self.cross_joins:
+            query &= table.id > 0
+
+        left_joins: list[Expression] = []
+        if names := self._required_join_names(joined, query, distinct):
+            self._build_left_joins_and_fields([], left_joins)
+        if left_joins := self._required_left_joins(names, left_joins):
+            inner_joins = self._build_inner_joins()
+            self._validate_joins(query, [], [model.id], {"join": inner_joins, "left": left_joins}, joined)
+            options: SelectKwargs = {"left": left_joins, "distinct": bool(distinct)}
+            if inner_joins:
+                options["join"] = inner_joins
+            counted = t.cast(Field, distinct if isinstance(distinct, (Field, Expression)) else model.id)
+            # COUNT(column) skips the NULL a left join yields, like COUNT(DISTINCT column) does:
+            subquery = db(query)._select(counted.with_alias("typedal_counted"), **options).rstrip().rstrip(";")
+            column = db._adapter.dialect.quote("typedal_counted")
+            return f"SELECT COUNT({column}) FROM ({subquery}) AS typedal_count;", distinct  # noqa: S608
+            # subquery is generated by pydal
+
         conditions: list[QueryLike] = []
         for key, relation in self.relationships.items():
             if not relation.condition:
@@ -1444,21 +1717,22 @@ def __count(
             if relation.join != "inner" and not include_left_join:
                 continue
 
-            other = relation.get_table(db)
-            if not distinct:
-                # todo: can this lead to other issues?
-                other = other.with_alias(f"{key}_{hash(relation)}")
+            # same alias as the select uses, so a filter on this relationship's table applies to this join:
+            other = relation.get_table(db).with_alias(f"{key}_{hash(relation)}")
 
             conditions.append(relation.condition(model, other))  # ty: ignore[invalid-argument-type]
             if callable(relation.condition_and):
                 conditions.append(relation.condition_and(model, other))  # ty: ignore[invalid-argument-type]
 
-        self._validate_joins(conditions, [model.id], {})
+        self._validate_joins(query, conditions, [model.id], {}, joined)
         for condition in conditions:
             query &= condition
-        for table in self.cross_joins:
-            query &= table.id > 0
-        return query
+        return query, distinct
+
+    def __execute_count(self, db: TypeDAL, query: Query | str, distinct: bool | Field | TypedField[t.Any]) -> int:
+        if isinstance(query, str):
+            return int(db.executesql(query)[0][0])
+        return db(query).count(distinct)
 
     def count(self, distinct: bool | Field | TypedField[t.Any] = False) -> int:
         """
@@ -1468,18 +1742,18 @@ def count(self, distinct: bool | Field | TypedField[t.Any] = False) -> int:
         """
         require_permission(self._permissions, "read")
         db = self._get_db()
-        query = self.__count(db, distinct=distinct)
+        query, distinct = self.__count(db, distinct=distinct)
 
-        return db(query).count(distinct)
+        return self.__execute_count(db, query, distinct)
 
     def _count(self, distinct: bool | Field | TypedField[t.Any] = False) -> str:
         """
         Return the SQL for .count().
         """
         db = self._get_db()
-        query = self.__count(db, distinct=distinct)
+        query, distinct = self.__count(db, distinct=distinct)
 
-        return db(query)._count(distinct)
+        return query if isinstance(query, str) else db(query)._count(distinct)
 
     def exists(self) -> bool:
         """
@@ -1499,8 +1773,8 @@ def __pagination_count(self) -> int:
 
         db = self._get_db()
         distinct = t.cast(TypedField[int], self.model.id)
-        query = self.__count(db, distinct=distinct, include_left_for_distinct=False)
-        return db(query).count(distinct)
+        query, distinct = self.__count(db, distinct=distinct, include_left_for_distinct=False)
+        return self.__execute_count(db, query, distinct)
 
     def __paginate(
         self,
diff --git a/tests/test_join_aliases.py b/tests/test_join_aliases.py
new file mode 100644
index 0000000..9caee50
--- /dev/null
+++ b/tests/test_join_aliases.py
@@ -0,0 +1,256 @@
+"""
+Joined relationships live under an alias; where() and orderby() on the joined table resolve to it when the builder runs.
+"""
+
+import typing as t
+
+import pytest
+
+from src.typedal import AliasedTableMismatchError, TypeDAL, TypedTable, relationship
+
+
+def define_models(db: TypeDAL):
+    @db.define()
+    class AliasAuthor(TypedTable):
+        name: str
+
+        articles = relationship(
+            list["AliasArticle"],
+            condition=lambda author, article: author.id == article.author,
+            join="left",
+        )
+
+    @db.define()
+    class AliasArticle(TypedTable):
+        title: str
+        author: AliasAuthor
+        reviewer: AliasAuthor
+
+        writer = relationship(AliasAuthor, condition=lambda article, author: article.author == author.id, join="left")
+        critic = relationship(AliasAuthor, condition=lambda article, author: article.reviewer == author.id, join="left")
+        comments = relationship(
+            list["AliasComment"],
+            condition=lambda article, comment: article.id == comment.article,
+            join="left",
+        )
+        # `on` joins the table itself, without an alias:
+        replies = relationship(
+            list["AliasComment"],
+            on=lambda article, comment: [comment.on(comment.article == article.id)],
+        )
+
+    @db.define()
+    class AliasComment(TypedTable):
+        article: AliasArticle
+        body: str
+
+        post = relationship(AliasArticle, condition=lambda comment, article: comment.article == article.id)
+
+    ann, bob, cat = AliasAuthor.bulk_insert([{"name": "ann"}, {"name": "bob"}, {"name": "cat"}])
+    first = AliasArticle.insert(title="First", author=ann, reviewer=bob)
+    AliasArticle.insert(title="Second", author=bob, reviewer=ann)
+    third = AliasArticle.insert(title="Third", author=ann, reviewer=cat)
+    AliasComment.bulk_insert(
+        [
+            {"article": first, "body": "nice"},
+            {"article": first, "body": "meh"},
+            {"article": third, "body": "nice"},
+        ]
+    )
+    db.commit()
+
+    return AliasAuthor, AliasArticle, AliasComment
+
+
+db = TypeDAL("sqlite:memory")
+Author, Article, Comment = define_models(db)
+
+
+def titles(builder) -> list[str]:
+    return [row.title for row in builder]
+
+
+def test_filter_on_left_join_paginates_and_counts():
+    builder = Article.join("comments").where(Comment.body == "nice").orderby(Article.id)
+
+    rows = builder.collect()
+    assert titles(rows) == ["First", "Third"]
+    # the filter applies to the joined rows too, like condition_and does:
+    assert [[comment.body for comment in row.comments] for row in rows] == [["nice"], ["nice"]]
+
+    # with limitby, the predicate moves into an id subquery, which needs the left join as well:
+    page = builder.paginate(limit=1)
+    assert titles(page) == ["First"]
+    assert (page.pagination["total_items"], page.pagination["total_pages"]) == (2, 2)
+    assert titles(builder.paginate(limit=1, page=2)) == ["Third"]
+
+    assert builder.count() == 2
+    assert builder.count(distinct=Article.id) == 2
+    assert "LEFT JOIN" in builder._count()
+
+    with_inner = Article.join("writer", method="inner").join("comments").where(Comment.body == "nice")
+    assert with_inner.count() == 2
+    assert with_inner.paginate(limit=1).pagination["total_items"] == 2
+
+
+def test_count_distinct_joined_field():
+    # "Second" has no comments, so its left-joined row is NULL, which isn't a distinct value:
+    assert Article.join("comments").count(distinct=Comment.body) == 2
+    assert Article.join("comments").where(Article.title != "Third").count(distinct=Comment.body) == 2
+
+
+def test_filter_for_missing_left_join_rows():
+    builder = Article.join("comments").where(Comment.id == None)
+
+    assert titles(builder) == ["Second"]
+    assert builder.paginate(limit=1).pagination["total_items"] == 1
+    assert builder.count() == 1
+
+
+def test_expression_on_joined_table_resolves():
+    builder = Article.join("comments", method="inner").where(Comment.body.upper() == "MEH")
+
+    assert titles(builder) == ["First"]
+    assert builder.count() == 1
+
+
+def test_orderby_on_joined_table_resolves():
+    builder = Article.join("writer", method="inner").orderby(~Author.name, Article.id)
+
+    assert titles(builder) == ["Second", "First", "Third"]
+    assert titles(builder.paginate(limit=2)) == ["Second", "First"]
+
+
+def test_same_table_joined_twice_needs_a_lambda():
+    builder = Article.join("writer").join("critic")
+
+    with pytest.raises(AliasedTableMismatchError, match="multiple aliases.*'writer', 'critic'"):
+        builder.where(Author.name == "ann").to_sql()
+
+    assert titles(builder.where(lambda article, critic: critic.name == "ann")) == ["Second"]
+    assert titles(builder.where(lambda article, writer: writer.name == "ann").orderby(Article.id)) == [
+        "First",
+        "Third",
+    ]
+    # the lambda is resolved when the builder runs, so it can come before the joins:
+    deferred = Article.where(lambda article, critic: critic.name == "ann").join("writer").join("critic")
+    assert titles(deferred) == ["Second"]
+    assert deferred.count() == 1
+    assert deferred.paginate(limit=1).pagination["total_items"] == 1
+
+
+def test_lambda_mixes_with_other_where_parts():
+    builder = Article.join("critic").where(
+        lambda article, critic: critic.name == "cat",
+        lambda article: article.title == "First",
+        title="Second",
+    )
+
+    assert titles(builder.orderby(Article.id)) == ["First", "Second", "Third"]
+
+    # like a plain lambda, one that returns nothing doesn't filter:
+    assert len(Article.join("critic").where(lambda article, critic: None).collect()) == 3
+
+    # an argument with a default doesn't ask for a joined table:
+    assert titles(Article.where(lambda article, title="Second": article.title == title)) == ["Second"]
+
+
+def test_lambda_asking_for_unjoined_relationship():
+    with pytest.raises(ValueError, match="unjoined relationship.*critic.*joined: none"):
+        Article.where(lambda article, critic: critic.name == "ann").to_sql()
+
+    with pytest.raises(ValueError, match=r"unjoined relationship\(s\) critic \(joined: writer\)"):
+        Article.join("writer").where(lambda article, critic: critic.name == "ann").collect()
+
+
+def test_nested_relationship_resolves():
+    builder = Author.join("articles.comments").where(Comment.body == "meh")
+
+    rows = builder.collect()
+    assert [row.name for row in rows] == ["ann"]
+    assert [article.title for article in rows.first().articles] == ["First"]
+    assert [comment.body for comment in rows.first().articles[0].comments] == ["meh"]
+    assert builder.paginate(limit=1).pagination["total_items"] == 1
+
+    by_path = Author.join("articles.comments").where(lambda author, articles__comments: articles__comments.body == "meh")
+    by_name = Author.join("articles.comments").where(lambda author, comments: comments.body == "meh")
+    assert [row.name for row in by_path] == [row.name for row in by_name] == ["ann"]
+
+
+def test_ambiguous_nested_name_needs_full_path():
+    builder = Comment.join("post.writer.articles", "post.critic.articles")
+
+    with pytest.raises(ValueError, match="unjoined relationship.*articles"):
+        builder.where(lambda comment, articles: articles.title == "Second").to_sql()
+
+    # comments on posts whose writer (ann) also wrote "Third":
+    by_writer = builder.where(lambda comment, post__writer__articles: post__writer__articles.title == "Third")
+    assert sorted(row.body for row in by_writer) == ["meh", "nice", "nice"]
+    # comments on posts whose critic (bob, cat) wrote "Second": only bob reviewed one with comments
+    by_critic = builder.where(lambda comment, post__critic__articles: post__critic__articles.title == "Second")
+    assert sorted(row.body for row in by_critic) == ["meh", "nice"]
+
+
+def test_on_relationship_keeps_the_table_itself():
+    replies = Article.join("replies").where(Comment.body == "meh")
+
+    assert titles(replies) == ["First"]
+    assert titles(replies.paginate(limit=1)) == ["First"]
+    assert replies.count() == 1
+
+    # joined both with and without an alias: the table name means the un-aliased `on` join
+    both = Article.join("replies", "comments").where(Comment.body == "meh")
+    assert titles(both) == ["First"]
+    assert sorted(comment.body for comment in both.first().comments) == ["meh", "nice"]
+
+
+def test_mutations_reject_lambdas_on_joined_tables():
+    builder = Article.join("critic").where(lambda article, critic: critic.name == "ann")
+
+    for mutation in (builder.delete, builder._delete):
+        with pytest.raises(ValueError, match=r"delete\(\) and update\(\)"):
+            mutation()
+    for mutation in (builder.update, builder._update):
+        with pytest.raises(ValueError, match=r"delete\(\) and update\(\)"):
+            mutation(title="changed")
+
+    assert Article.count() == 3
+
+
+def test_cache_key_does_not_depend_on_alias_hashes():
+    def build():
+        # every join() clones the relationship, so each builder gets its own alias hashes:
+        return Article.join("critic").where(lambda article, critic: critic.name == "ann")
+
+    first, second = build(), build()
+    assert hash(first.relationships["critic"]) != hash(second.relationships["critic"])
+
+    key = first._cache_key_query()
+    assert key == second._cache_key_query()
+    assert "<critic>" in t.cast(str, key)
+    assert str(hash(first.relationships["critic"])) not in t.cast(str, key)
+
+    # the un-aliased `on` join keeps its table name:
+    with_on = (
+        Article.join("replies", "critic")
+        .where(lambda article, critic: critic.name == "ann")
+        .where(Comment.body == "meh")
+    )
+    with_on_key = t.cast(str, with_on._cache_key_query())
+    assert '"alias_comment"."body"' in with_on_key
+    assert "<critic>" in with_on_key
+
+    plain = Article.where(title="First")
+    assert plain._cache_key_query() is plain.query
+
+
+def test_left_join_filter_paginates_with_distinct_psql(dal_psql: TypeDAL):
+    _, article, comment = define_models(dal_psql)
+
+    builder = article.join("comments").where(comment.body == "nice")
+    assert titles(builder.paginate(limit=1)) == ["First"]
+    assert builder.paginate(limit=1).pagination["total_items"] == 2
+
+    ordered = builder.orderby(~comment.id)
+    assert titles(ordered.paginate(limit=1)) == ["Third"]
+    assert ordered.count(distinct=article.id) == 2
diff --git a/tests/test_query_builder.py b/tests/test_query_builder.py
index b142a39..8022344 100644
--- a/tests/test_query_builder.py
+++ b/tests/test_query_builder.py
@@ -3,7 +3,6 @@
 import pytest
 
 from src.typedal import (
-    AliasedTableMismatchError,
     ImplicitCrossJoinError,
     QueryBuilder,
     TypeDAL,
@@ -88,7 +87,7 @@ def _setup_data():
     db.commit()
 
 
-def test_where_on_unaliased_joined_table_is_rejected():
+def test_where_on_joined_table_resolves_to_its_alias():
     _setup_data()
     safe = TestQueryTable.join(
         "relations",
@@ -101,14 +100,16 @@ def test_where_on_unaliased_joined_table_is_rejected():
     assert [row.number for row in rows] == [1]
     assert [relation.value for relation in rows.first().relations] == [33, 33, 33, 33]
 
-    unsafe = (
-        TestQueryTable.join("relations", method="inner")
-        .select(TestRelationship.name)
-        .where(TestRelationship.value > 0)
-    )
-    with pytest.raises(AliasedTableMismatchError, match="alias"):
-        unsafe.to_sql()
-
+    # a builder that is extended later can't use condition_and anymore, so a plain where() on the joined table
+    # targets the alias too, whether it comes before or after the join:
+    joined_first = TestQueryTable.join("relations", method="inner").where(TestRelationship.value > 10)
+    where_first = TestQueryTable.where(TestRelationship.value > 10).join("relations", method="inner")
+    for builder in (joined_first, where_first):
+        assert "CROSS JOIN" not in builder.to_sql()
+        assert [row.number for row in builder.collect()] == [1]
+        assert builder.count() == 4
+        assert builder.count(distinct=TestQueryTable.id) == 1
+        assert builder.paginate(limit=1).pagination["total_items"] == 1
 
 def test_where_on_unrelated_table_is_rejected():
     with pytest.raises(ImplicitCrossJoinError, match="cross join"):

From 6ed8d56902e9534ae51f6ab457294f782c4442f1 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Fri, 2 Oct 2026 10:39:10 +0000
Subject: [PATCH 2/4] fix(query-builder): nested joins back to the root table
 and across rows

- A nested relationship to the root table (`Article.join("writer.articles")`)
  renamed the root's own select fields to the nested alias, so collecting
  raised KeyError. It is now aliased upfront like a self-reference.
- Nested relationships lost every row after the first one: once a parent
  instance was seen, later rows (e.g. an article's second comment) were
  skipped before their nested data was read. Rows reached through
  different parents also shared one dedup set keyed by the nested row's
  id. Dedup is now per parent instance, and a seen instance still takes
  nested data from later rows.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FXZN6UgVPuRxvHzMMTShAE
---
 src/typedal/query_builder.py | 45 +++++++++++++++++++-----------------
 tests/test_join_aliases.py   | 44 ++++++++++++++++++++++++++++++++++-
 2 files changed, 67 insertions(+), 22 deletions(-)

diff --git a/src/typedal/query_builder.py b/src/typedal/query_builder.py
index eed4a73..5b3ace1 100644
--- a/src/typedal/query_builder.py
+++ b/src/typedal/query_builder.py
@@ -102,6 +102,10 @@ def _walk_tables(node: object, tables: set[str], parents: dict[str, str]) -> set
     return found
 
 
+# (id of the parent record, relationship path, related row id) -> the related instance attached to that parent
+type SeenRelations = dict[tuple[int, str, t.Any], t.Any]
+
+
 class _JoinedTable(t.NamedTuple):
     """A relationship table as it ends up in the SQL of a builder with joins."""
 
@@ -764,9 +768,9 @@ def _collect_joined_tables(
         joined: list[_JoinedTable],
     ) -> None:
         # mirrors the aliasing in _process_relationship_for_left_join and _build_inner_joins_recursive:
-        # everything is aliased, except an `on` relationship to another table.
+        # everything is aliased, except an `on` relationship to another table than its parent or the root.
         other = relation.get_table(self._get_db())
-        aliased = other is parent or not relation.on
+        aliased = other is parent or other is self.model or not relation.on
         table = other.with_alias(f"{key}_{hash(relation)}") if aliased else other
         joined.append(_JoinedTable(path, relation, t.cast(Table, table), other._tablename, aliased))
 
@@ -1363,9 +1367,10 @@ def _process_relationship_for_left_join(
 
         select_fields = ", ".join([str(_) for _ in select_args])
         pre_alias = str(other)
-        # Self-referencing relationship: 'other' is the same table as 'parent_table', so the
-        # name-based helpers below can't tell their fields apart. Alias upfront and skip them.
-        is_self_reference = other is parent_table
+        # Self-referencing relationship: 'other' is the same table as 'parent_table' (or, for a nested relationship
+        # like 'writer.articles', as the root table), so the name-based helpers below can't tell their fields apart.
+        # Alias upfront and skip them.
+        is_self_reference = other is parent_table or other is self.model
 
         if is_self_reference:
             other = other.with_alias(f"{key}_{hash(relation)}")
@@ -1476,8 +1481,8 @@ def _collect_with_relationships(
         # id: [Row]
         raw_per_id: dict[t.Any, list[t.Any]] = defaultdict(list)
 
-        # Track what we've seen: main_id -> "column-relation_id"
-        seen_relations: dict[str, set[str]] = defaultdict(set)
+        # Track what we've seen: (parent record, column, relation id) -> instance
+        seen_relations: SeenRelations = {}
 
         for row in rows:
             main = row[main_table]
@@ -1502,7 +1507,6 @@ def _collect_with_relationships(
                     column=column,
                     relation=relation,
                     parent_record=records[main_id],
-                    parent_id=main_id,
                     seen_relations=seen_relations,
                     db=db,
                 )
@@ -1515,8 +1519,7 @@ def _process_relationship_data(
         column: str,
         relation: Relationship[t.Any],
         parent_record: t.Any,
-        parent_id: t.Any,
-        seen_relations: dict[str, set[str]],
+        seen_relations: SeenRelations,
         db: t.Any,
         path: str = "",
     ) -> t.Any | None:
@@ -1530,13 +1533,12 @@ def _process_relationship_data(
             column: The relationship column name
             relation: The Relationship object
             parent_record: The parent model instance to attach data to
-            parent_id: ID of the parent for tracking
             seen_relations: Dict tracking which relationships we've already processed
             db: Database instance
             path: Current relationship path (e.g., "users.bestie")
 
         Returns:
-            The created relationship instance, or None if skipped
+            The created relationship instance, or None if skipped (no data, or already attached)
         """
         # Build the full path for tracking (e.g., "users", "users.bestie", "users.bestie.articles")
         current_path = f"{path}.{column}" if path else column
@@ -1551,16 +1553,19 @@ def _process_relationship_data(
         if not relation_data or relation_data.id is None:
             return None
 
-        # Check if we've already seen this relationship instance
-        seen_key = f"{current_path}-{relation_data.id}"
-        if seen_key in seen_relations[parent_id]:
-            return None  # Already processed
-
-        seen_relations[parent_id].add(seen_key)
+        # Check if we've already attached this relationship instance to this parent. Keyed by the parent object:
+        # the same row (e.g. an author) reached through different parents is a separate instance for each of them.
+        seen_key = (id(parent_record), current_path, relation_data.id)
+        if (seen := seen_relations.get(seen_key)) is not None:
+            # a later row can still hold new data for its nested relationships (e.g. the author's second article)
+            if relation.nested:
+                self._process_nested_relationships(row, relation, seen, seen_relations, db, current_path)
+            return None
 
         # Create the relationship instance
         relation_table = relation.get_table(db)
         instance = relation_table(relation_data) if looks_like(relation_table, TypedTable) else relation_data
+        seen_relations[seen_key] = instance
 
         # Process nested relationships on this instance
         if relation.nested:
@@ -1568,7 +1573,6 @@ def _process_relationship_data(
                 row=row,
                 relation=relation,
                 instance=instance,
-                # parent_id=parent_id,
                 seen_relations=seen_relations,
                 db=db,
                 path=current_path,
@@ -1590,7 +1594,7 @@ def _process_nested_relationships(
         row: t.Any,
         relation: Relationship[t.Any],
         instance: t.Any,
-        seen_relations: dict[str, set[str]],
+        seen_relations: SeenRelations,
         db: t.Any,
         path: str,
     ) -> None:
@@ -1618,7 +1622,6 @@ def _process_nested_relationships(
                 column=nested_col,
                 relation=nested_relation,
                 parent_record=instance,
-                parent_id=instance.id,
                 seen_relations=seen_relations,
                 db=db,
                 path=path,
diff --git a/tests/test_join_aliases.py b/tests/test_join_aliases.py
index 9caee50..cacc554 100644
--- a/tests/test_join_aliases.py
+++ b/tests/test_join_aliases.py
@@ -172,11 +172,53 @@ def test_nested_relationship_resolves():
     assert [comment.body for comment in rows.first().articles[0].comments] == ["meh"]
     assert builder.paginate(limit=1).pagination["total_items"] == 1
 
-    by_path = Author.join("articles.comments").where(lambda author, articles__comments: articles__comments.body == "meh")
+    by_path = Author.join("articles.comments").where(
+        lambda author, articles__comments: articles__comments.body == "meh"
+    )
     by_name = Author.join("articles.comments").where(lambda author, comments: comments.body == "meh")
     assert [row.name for row in by_path] == [row.name for row in by_name] == ["ann"]
 
 
+def test_nested_one_to_many_collects_every_row():
+    # "First" has two comments, which arrive on two rows that repeat ann and "First":
+    authors = Author.join("articles.comments").orderby(Author.id).collect()
+
+    assert [
+        (author.name, [(article.title, sorted(c.body for c in article.comments)) for article in author.articles])
+        for author in authors
+    ] == [
+        ("ann", [("First", ["meh", "nice"]), ("Third", ["nice"])]),
+        ("bob", [("Second", [])]),
+        ("cat", []),
+    ]
+
+
+def test_nested_relationship_back_to_the_root_table():
+    def overview(builder):
+        return [(row.title, row.writer.name, sorted(a.title for a in row.writer.articles)) for row in builder]
+
+    expected = [
+        ("First", "ann", ["First", "Third"]),
+        ("Second", "bob", ["Second"]),
+        ("Third", "ann", ["First", "Third"]),
+    ]
+
+    assert overview(Article.join("writer.articles").orderby(Article.id)) == expected
+    assert overview(Article.join("writer.articles", method="inner").orderby(Article.id)) == expected
+
+    # a selected root field stays the root's, it isn't moved to the nested 'articles' alias:
+    selected = Article.join("writer.articles").select(Article.title).orderby(Article.id).collect()
+    assert [row.title for row in selected] == ["First", "Second", "Third"]
+    assert [len(row.writer.articles) for row in selected] == [2, 1, 2]
+
+    # and a filter on the root table keeps meaning the root:
+    assert overview(Article.join("writer.articles").where(Article.title == "Second")) == [("Second", "bob", ["Second"])]
+    assert (
+        Article.join("writer.articles").where(Article.title == "Second").paginate(limit=1).pagination["total_items"]
+        == 1
+    )
+
+
 def test_ambiguous_nested_name_needs_full_path():
     builder = Comment.join("post.writer.articles", "post.critic.articles")
 

From dfd85ae8f08275fd7d2af9bc6911cc25c5388ccc Mon Sep 17 00:00:00 2001
From: Robin van der Noord <robinvandernoord@gmail.com>
Date: Fri, 2 Oct 2026 14:40:31 +0200
Subject: [PATCH 3/4] fix(query-builder): preserve join filters in counts and
 pagination

---
 src/typedal/query_builder.py | 44 +++++++++++++++++++------
 tests/test_join_aliases.py   | 63 ++++++++++++++++++++++++++++++++++++
 2 files changed, 98 insertions(+), 9 deletions(-)

diff --git a/src/typedal/query_builder.py b/src/typedal/query_builder.py
index 5b3ace1..6aae7cf 100644
--- a/src/typedal/query_builder.py
+++ b/src/typedal/query_builder.py
@@ -86,6 +86,11 @@ def _walk_tables(node: object, tables: set[str], parents: dict[str, str]) -> set
     Module-level rather than a closure: a self-referencing nested function creates a reference cycle per call,
     and the extra cyclic GC runs made SQL building noticeably slower.
     """
+    if isinstance(node, (list, tuple, set)):
+        found: set[str] = set()
+        for item in node:
+            found |= _walk_tables(item, tables, parents)
+        return found
     if isinstance(node, Field):
         name = t.cast(str, node.tablename)
         tables.add(name)
@@ -1178,7 +1183,10 @@ def _handle_relationships_pre_select(
             {"join": inner_joins or select_kwargs.get("join"), "left": left_joins},
             joined,
         )
-        required_left = self._required_left_joins(self._required_join_names(joined, query), left_joins)
+        required_names = self._required_join_names(
+            joined, query, select_kwargs.get("orderby"), select_kwargs.get("groupby"), select_kwargs.get("having")
+        )
+        required_left = self._required_left_joins(required_names, left_joins)
         query = self._apply_limitby_optimization(query, select_kwargs, inner_joins, metadata, required_left)
 
         if inner_joins:
@@ -1293,10 +1301,28 @@ def _select_distinct_ids_with_orderby_fields(self, query: Query, select_kwargs:
                 select_args.append(field)
                 seen.add(key)
 
-        ids = db(query)._select(*select_args, **select_kwargs).rstrip(";")
-        id_column = db._adapter.dialect.quote("typedal_paginate_id")
-        return f"SELECT {id_column} FROM ({ids}) AS typedal_paginate_ids"  # noqa: S608
-        # ids is a safe subquery, originating from code
+        kwargs = select_kwargs.copy()
+        limitby = kwargs.pop("limitby", None)
+        orderby = kwargs.pop("orderby")
+        ordering = orderby if isinstance(orderby, (list, tuple)) else [orderby]
+        sql_orderby = ", ".join(db._adapter.expand(field) for field in ordering)
+        # Rank joined rows before reducing them to one entry per root. The first
+        # occurrence in the requested ordering determines each root's page.
+        select_args.append(f"ROW_NUMBER() OVER (ORDER BY {sql_orderby}) AS typedal_paginate_position")
+        kwargs["distinct"] = False
+        ids = db(query)._select(*select_args, **kwargs).rstrip(";")
+        dialect = db._adapter.dialect
+        id_column = dialect.quote("typedal_paginate_id")
+        position_column = dialect.quote("typedal_paginate_position")
+        limited_ids = dialect.select(
+            id_column,
+            f"({ids}) AS typedal_paginate_ids",
+            groupby=id_column,
+            orderby=f"MIN({position_column})",
+            limitby=limitby,
+        ).rstrip(";")
+        # MySQL requires an unlimited outer SELECT for a limited IN subquery.
+        return dialect.select(id_column, f"({limited_ids}) AS typedal_paginate_limited_ids")
 
     def _apply_limitby_optimization(
         self,
@@ -1309,7 +1335,7 @@ def _apply_limitby_optimization(
         """
         Apply limitby optimization when relationships are present.
 
-        `left_joins` are the left joins the predicate filters on, see _required_left_joins.
+        `left_joins` are the left joins used by filtering and ordering, see _required_left_joins.
         """
         if not (limitby := select_kwargs.pop("limitby", ())):
             return query
@@ -1337,7 +1363,7 @@ def _apply_limitby_optimization(
 
         id_field = t.cast(TypedField[int], model.id)
 
-        query = id_field.belongs(ids)
+        query &= id_field.belongs(ids)
         metadata["ids"] = ids
 
         return query
@@ -1711,13 +1737,13 @@ def __count(
             return f"SELECT COUNT({column}) FROM ({subquery}) AS typedal_count;", distinct  # noqa: S608
             # subquery is generated by pydal
 
-        conditions: list[QueryLike] = []
+        conditions: list[QueryLike] = [join.second for join in self._build_inner_joins()]
         for key, relation in self.relationships.items():
             if not relation.condition:
                 continue
 
             include_left_join = distinct and include_left_for_distinct
-            if relation.join != "inner" and not include_left_join:
+            if relation.join == "inner" or not include_left_join:
                 continue
 
             # same alias as the select uses, so a filter on this relationship's table applies to this join:
diff --git a/tests/test_join_aliases.py b/tests/test_join_aliases.py
index cacc554..31380f8 100644
--- a/tests/test_join_aliases.py
+++ b/tests/test_join_aliases.py
@@ -99,6 +99,69 @@ def test_count_distinct_joined_field():
     assert Article.join("comments").where(Article.title != "Third").count(distinct=Comment.body) == 2
 
 
+def test_count_distinct_roots_applies_left_join_condition_and():
+    """Distinct root counts include only comments matching the additional join condition."""
+    builder = Article.join("comments", condition_and=lambda _article, comment: comment.body == "meh")
+
+    rows = builder.orderby(Article.id).collect()
+    assert titles(rows) == ["First", "Second", "Third"]
+    assert [[comment.body for comment in row.comments] for row in rows] == [["meh"], [], []]
+    assert builder.count(distinct=True) == 1
+
+
+def test_paginated_join_keeps_filtered_comments():
+    builder = Article.join("comments").where(Comment.body == "nice").orderby(Article.id)
+
+    assert [[comment.body for comment in row.comments] for row in builder.collect()] == [["nice"], ["nice"]]
+
+    for page_number, expected_title in enumerate(["First", "Third"], start=1):
+        page = builder.paginate(limit=1, page=page_number)
+        assert titles(page) == [expected_title]
+        assert [comment.body for comment in page.first().comments] == ["nice"]
+
+
+def test_paginated_left_join_includes_orderby_relationship():
+    builder = Article.join("writer").orderby(~Author.name, Article.id)
+
+    assert titles(builder.collect()) == ["Second", "First", "Third"]
+    assert [titles(builder.paginate(limit=1, page=page)) for page in range(1, 4)] == [
+        ["Second"],
+        ["First"],
+        ["Third"],
+    ]
+
+
+@pytest.mark.parametrize("operation", ["count", "paginate"])
+def test_nested_inner_join_filter_counts_and_paginates(operation):
+    builder = Author.join("articles.comments", method="inner").where(Comment.body == "meh")
+
+    rows = builder.collect()
+    assert [row.name for row in rows] == ["ann"]
+    assert [article.title for article in rows.first().articles] == ["First"]
+    assert [comment.body for comment in rows.first().articles[0].comments] == ["meh"]
+
+    if operation == "count":
+        assert builder.count() == 1
+    else:
+        page = builder.paginate(limit=1)
+        assert [row.name for row in page] == ["ann"]
+        assert (page.pagination["total_items"], page.pagination["total_pages"]) == (1, 1)
+
+
+def test_paginated_join_deduplicates_root_ids_before_limit():
+    builder = Article.join("comments").where(Comment.id > 0).orderby(Comment.id)
+
+    assert titles(builder.collect()) == ["First", "Third"]
+    assert builder.count(distinct=Article.id) == 2
+
+    first_page = builder.paginate(limit=1)
+    second_page = builder.paginate(limit=1, page=2)
+
+    assert (first_page.pagination["total_items"], first_page.pagination["total_pages"]) == (2, 2)
+    assert titles(first_page) == ["First"]
+    assert titles(second_page) == ["Third"]
+
+
 def test_filter_for_missing_left_join_rows():
     builder = Article.join("comments").where(Comment.id == None)
 

From e20f0d8c9d18be8e833394b0da12e69ed63a73fd Mon Sep 17 00:00:00 2001
From: Robin van der Noord <robinvandernoord@gmail.com>
Date: Fri, 2 Oct 2026 17:59:43 +0200
Subject: [PATCH 4/4] fix(query-builder): include transitive left-join
 dependencies

---
 src/typedal/query_builder.py | 17 +++++++++++--
 tests/test_join_aliases.py   | 48 ++++++++++++++++++++++++++++++++++++
 2 files changed, 63 insertions(+), 2 deletions(-)

diff --git a/src/typedal/query_builder.py b/src/typedal/query_builder.py
index 6aae7cf..b9aad00 100644
--- a/src/typedal/query_builder.py
+++ b/src/typedal/query_builder.py
@@ -844,12 +844,25 @@ def _required_join_names(joined: list[_JoinedTable], *nodes: t.Any) -> set[str]:
     @staticmethod
     def _required_left_joins(names: set[str], left_joins: list[Expression]) -> list[Expression]:
         """
-        The left joins of the tables in `names`, see _required_join_names.
+        The left joins of the tables in `names` and their transitive ON dependencies.
 
         Queries that replace the select (the id subquery of limitby, count) need these: without them, a filter on a
         left-joined table references a table that isn't in their FROM clause.
         """
-        return [join for join in left_joins if isinstance(join.first, Table) and join.first._tablename in names]
+        dependencies: dict[str, set[str]] = defaultdict(set)
+        for join in left_joins:
+            if isinstance(join.first, Table):
+                dependencies[join.first._tablename] |= _walk_tables(join.second, set(), {})
+
+        required = names.copy()
+        pending = names.copy()
+        while pending:
+            name = pending.pop()
+            referenced = dependencies.get(name, set()) - required
+            required |= referenced
+            pending |= referenced
+
+        return [join for join in left_joins if isinstance(join.first, Table) and join.first._tablename in required]
 
     def _before_query(self, mut_metadata: Metadata, add_id: bool = True) -> tuple[Query, list[t.Any], SelectKwargs]:
         select_args = [self._select_arg_convert(_) for _ in self.select_args] or [self.model.ALL]
diff --git a/tests/test_join_aliases.py b/tests/test_join_aliases.py
index 31380f8..157c1e2 100644
--- a/tests/test_join_aliases.py
+++ b/tests/test_join_aliases.py
@@ -148,6 +148,54 @@ def test_nested_inner_join_filter_counts_and_paginates(operation):
         assert (page.pagination["total_items"], page.pagination["total_pages"]) == (1, 1)
 
 
+@pytest.mark.parametrize("operation", ["count", "paginate"])
+def test_custom_on_intermediate_join_filter_counts_and_paginates(operation):
+    db = TypeDAL("sqlite:memory")
+
+    try:
+        @db.define()
+        class Parent(TypedTable):
+            name: str
+
+            children = relationship(
+                list["Child"],
+                on=lambda parent, child: [
+                    ParentChild.on(ParentChild.parent == parent.id),
+                    child.on(child.id == ParentChild.child),
+                ],
+                join="left",
+            )
+
+        @db.define()
+        class Child(TypedTable):
+            name: str
+
+        @db.define()
+        class ParentChild(TypedTable):
+            parent: Parent
+            child: Child
+
+        Parent.insert(name="unlinked")
+        linked = Parent.insert(name="linked")
+        child = Child.insert(name="matching")
+        ParentChild.insert(parent=linked, child=child)
+
+        builder = Parent.join("children").where(Child.name == "matching").orderby(Parent.id)
+        rows = builder.collect()
+        assert [row.id for row in rows] == [linked.id]
+        assert [child.id for child in rows.first().children] == [child.id]
+
+        if operation == "count":
+            assert builder.count() == 1
+            assert builder.count(distinct=Parent.id) == 1
+        else:
+            page = builder.paginate(limit=1)
+            assert [row.id for row in page] == [linked.id]
+            assert (page.pagination["total_items"], page.pagination["total_pages"]) == (1, 1)
+    finally:
+        db.close()
+
+
 def test_paginated_join_deduplicates_root_ids_before_limit():
     builder = Article.join("comments").where(Comment.id > 0).orderby(Comment.id)