TML-2785: M:N slice 1 — correlated include read through the junction (#679)

Slice 1 of the **SQL ORM: Many-to-Many End to End** project ([Linear
project](https://linear.app/prisma-company/project/sql-orm-many-to-many-end-to-end-c178df40ca3a)).
Reads an M:N relation through its junction.

> **Stacked PR.** Base is `tml-2784` (#678, slice 0) → `tml-2597` (#673)
→ `tml-2729` (#667) → `main`. Review/merge bottom-up.

## Overview

`db.orm.User.include(tags)` now resolves a many-to-many relation to `{
…user, tags: Tag[] }` in a **single correlated subquery** that walks
parent → junction → target — no `LATERAL`, no second query. Built on
slice 0s `ResolvedRelation.through`.

## Changes (4 commits)

- **`fcecac5b3`** — integration fixture gains a `User ↔ Tag` M:N
relation via a `UserTag` junction (composite PK `user_id`/`tag_id`);
`contract.json`/`.d.ts` re-emitted.
- **`e587b433c`** — read path: `IncludeExpr.through` (surfaced by
`resolveIncludeRelation`), and `buildCorrelatedIncludeProjection` gains
an M:N branch — `buildManyToManyJunctionArtifacts` builds a non-LATERAL
inner join to the junction (`junction.childColumns =
target.targetColumns`) correlated to the parent (`junction.parentColumns
= parent` anchor), composite-key AND-ed; FK decode path reused.
Unit-tested at the AST level.
- **`b9c3e9f7b`** — replace 2 bare `as` casts with `castAs`; add the
missing M:N + distinct + non-leaf unit test.
- **`d3232cbad`** — 7 integration tests (PGlite).

## Integration tests (per the project standard)

Whole-row `toEqual`; 6/7 use explicit `.select(...)` (so adding a model
field wont churn assertions); **test 5 uses implicit/default selection**
(full `User` + `tags: Tag[]` shape); a **single-execution /
no-`LATERAL`** assertion; depth-2 nesting (`invitedUsers → tags`); edges
(user with no tags → `tags: []`; a tag shared by multiple users).

## Why

This is the first of the three relation-shaped M:N consumers (read /
filter / write) over slice 0s shared `through` primitive. The
correlated-only approach matches the post-TML-2729 read path (no LATERAL
to reintroduce).

## Scope / notes

Read only — filter (TML-2786) and write (TML-2787) are later slices. The
fixture is **one-directional** (`User.tags`; reverse `Tag.users`
deferred — adding it trips a latent create-overload type fragility in
unrelated mutation-defaults tests; see the projects unattended-decisions
log). Fixture re-emit used a `tsx` bypass because the CLI `contract
emit` fails on a sandbox config-load env issue — **CI `fixtures:check`
is the real golden-stability gate**; please confirm its green (or re-run
the canonical emit). Broad integration runs show pre-existing
PGlite/WASM JIT flakiness; the M:N tests pass on targeted runs.

Refs: TML-2785.

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* include(...) now supports many-to-many (M:N) relationships via
junction tables, returning correct nested arrays and preserving
single-query execution.

* **Tests**
* Added unit and integration tests covering M:N includes, composite
keys, nested includes, distinct+nested scenarios, and end-to-end result
shape correctness.

* **Documentation**
* Added upgrade notes for 0.14 describing the runtime support for M:N
correlated include reads.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Signed-off-by: Alexey Orlenko's AI Agent <robot@aqrln.net>

Author: tensordreams

packages/3-extensions/sql-orm-client/src/collection-contract.ts

@@ -6,12 +6,13 @@ import type {
 } from '@prisma-next/contract/types';
 import type { SqlStorage, StorageTable } from '@prisma-next/sql-contract/types';
 import { blindCast } from '@prisma-next/utils/casts';
+import { ifDefined } from '@prisma-next/utils/defined';
 import {
   domainModelTableInNamespace,
   resolveTableForContract,
   storageTableForContract,
 } from './storage-resolution';
-import type { RelationCardinalityTag } from './types';
+import type { IncludeThroughDescriptor, RelationCardinalityTag } from './types';
 
 type ModelStorageFields = Record<string, { column?: string }>;
 type ModelEntry = {
@@ -297,6 +298,7 @@ export interface ResolvedIncludeRelation {
   readonly targetColumn: string;
   readonly localColumn: string;
   readonly cardinality: RelationCardinalityTag | undefined;
+  readonly through?: IncludeThroughDescriptor;
 }
 
 export function resolveIncludeRelation(
@@ -327,13 +329,29 @@ export function resolveIncludeRelation(
     targetField,
   );
 
+  let through: IncludeThroughDescriptor | undefined;
+  if (relation.through !== undefined) {
+    const parentLocalColumns = relation.on.localFields.map((field) =>
+      resolveFieldToColumn(contract, namespaceId, modelName, field),
+    );
+    through = {
+      table: relation.through.table,
+      namespaceId: relation.through.namespaceId,
+      parentColumns: relation.through.parentColumns,
+      childColumns: relation.through.childColumns,
+      targetColumns: relation.through.targetColumns,
+      parentLocalColumns,
+    };
+  }
+
   return {
     relatedModelName: relation.to,
     relatedNamespaceId: relation.toNamespace,
     relatedTableName,
     targetColumn,
     localColumn,
     cardinality: relation.cardinality,
+    ...ifDefined('through', through),
   };
 }
 

packages/3-extensions/sql-orm-client/src/collection-dispatch.ts

@@ -100,7 +100,9 @@ function dispatchWithIncludes<Row>(options: {
   const generator = async function* (): AsyncGenerator<Row, void, unknown> {
     const { scope, release } = await acquireRuntimeScope(runtime);
     try {
-      const parentJoinColumns = state.includes.map((include) => include.localColumn);
+      const parentJoinColumns = state.includes.flatMap((include) =>
+        include.through !== undefined ? include.through.parentLocalColumns : [include.localColumn],
+      );
       const { selectedForQuery: parentSelectedForQuery, hiddenColumns: hiddenParentColumns } =
         augmentSelectionForJoinColumns(state.selectedFields, parentJoinColumns);
       const compiled = compileSelectWithIncludes(

packages/3-extensions/sql-orm-client/src/collection.ts

@@ -16,6 +16,7 @@ import {
   type ToWhereExpr,
   type WhereArg,
 } from '@prisma-next/sql-relational-core/ast';
+import { ifDefined } from '@prisma-next/utils/defined';
 import type { SimplifyDeep } from '@prisma-next/utils/simplify-deep';
 import { createAggregateBuilder, isAggregateSelector } from './aggregate-builder';
 import { normalizeAggregateResult } from './collection-aggregate-result';
@@ -513,6 +514,7 @@ export class Collection<
       targetColumn: relation.targetColumn,
       localColumn: relation.localColumn,
       cardinality: relation.cardinality,
+      ...ifDefined('through', relation.through),
       nested: nestedState,
       scalar: scalarSelector,
       combine: combineBranches,

packages/3-extensions/sql-orm-client/src/query-plan-select.ts

@@ -25,6 +25,7 @@ import {
 } from '@prisma-next/sql-relational-core/ast';
 import { codecRefForStorageColumn } from '@prisma-next/sql-relational-core/codec-descriptor-registry';
 import type { SqlQueryPlan } from '@prisma-next/sql-relational-core/plan';
+import { assertDefined, invariant } from '@prisma-next/utils/assertions';
 import { ifDefined } from '@prisma-next/utils/defined';
 import {
   type PolymorphismInfo,
@@ -341,6 +342,79 @@ function buildChildPolymorphismJoinsAndProjection(
   };
 }
 
+/**
+ * Build the correlated WHERE and junction JOIN artifacts for a many-to-many
+ * include. The resulting WHERE correlates the junction to the parent rows
+ * (AND-ed across all column pairs for composite keys). The junction JOIN
+ * connects child rows to the junction via the child columns.
+ */
+function buildManyToManyJunctionArtifacts(
+  parentTableName: string,
+  childTableRef: string,
+  through: NonNullable<IncludeExpr['through']>,
+): {
+  readonly whereExpr: AnyExpression;
+  readonly junctionJoin: JoinAst;
+} {
+  const {
+    table: junctionTable,
+    parentColumns,
+    childColumns,
+    targetColumns,
+    parentLocalColumns,
+    namespaceId,
+  } = through;
+
+  invariant(
+    childColumns.length === targetColumns.length,
+    `M:N junction '${junctionTable}': childColumns (${childColumns.length}) and targetColumns (${targetColumns.length}) must have equal length`,
+  );
+  invariant(
+    parentColumns.length === parentLocalColumns.length,
+    `M:N junction '${junctionTable}': parentColumns (${parentColumns.length}) and parentLocalColumns (${parentLocalColumns.length}) must have equal length`,
+  );
+
+  const joinOnPairs = childColumns.map((junctionCol, i) => {
+    const targetCol = targetColumns[i];
+    assertDefined(
+      targetCol,
+      `M:N junction '${junctionTable}': missing target column at index ${i}`,
+    );
+    return BinaryExpr.eq(
+      ColumnRef.of(junctionTable, junctionCol),
+      ColumnRef.of(childTableRef, targetCol),
+    );
+  });
+  const firstJoinPair = joinOnPairs[0];
+  const joinOn: AnyExpression =
+    joinOnPairs.length === 1 && firstJoinPair ? firstJoinPair : AndExpr.of(joinOnPairs);
+
+  const correlationPairs = parentColumns.map((junctionCol, i) => {
+    const parentLocalCol = parentLocalColumns[i];
+    assertDefined(
+      parentLocalCol,
+      `M:N junction '${junctionTable}': missing parent-local column at index ${i}`,
+    );
+    return BinaryExpr.eq(
+      ColumnRef.of(junctionTable, junctionCol),
+      ColumnRef.of(parentTableName, parentLocalCol),
+    );
+  });
+  const firstCorrelationPair = correlationPairs[0];
+  const whereExpr: AnyExpression =
+    correlationPairs.length === 1 && firstCorrelationPair
+      ? firstCorrelationPair
+      : AndExpr.of(correlationPairs);
+
+  const junctionJoin = JoinAst.inner(
+    TableSource.named(junctionTable, undefined, namespaceId),
+    joinOn,
+    false,
+  );
+
+  return { whereExpr, junctionJoin };
+}
+
 function buildIncludeChildRowsSelect(
   contract: Contract<SqlStorage>,
   parentTableName: string,
@@ -377,11 +451,25 @@ function buildIncludeChildRowsSelect(
     filterTableName: include.relatedTableName,
     namespaceId: include.relatedNamespaceId,
   });
-  const joinExpr = BinaryExpr.eq(
-    ColumnRef.of(childTableRef, include.targetColumn),
-    ColumnRef.of(parentTableName, include.localColumn),
-  );
-  const whereExpr = childWhere ? AndExpr.of([joinExpr, childWhere]) : joinExpr;
+
+  let whereExpr: AnyExpression;
+  let junctionJoins: JoinAst[] = [];
+
+  if (include.through !== undefined) {
+    const artifacts = buildManyToManyJunctionArtifacts(
+      parentTableName,
+      childTableRef,
+      include.through,
+    );
+    whereExpr = childWhere ? AndExpr.of([artifacts.whereExpr, childWhere]) : artifacts.whereExpr;
+    junctionJoins = [artifacts.junctionJoin];
+  } else {
+    const joinExpr = BinaryExpr.eq(
+      ColumnRef.of(childTableRef, include.targetColumn),
+      ColumnRef.of(parentTableName, include.localColumn),
+    );
+    whereExpr = childWhere ? AndExpr.of([joinExpr, childWhere]) : joinExpr;
+  }
 
   // `distinct()` on a non-leaf include cannot be lowered as
   // `SELECT DISTINCT <scalars>, json_agg(<grandchild>) FROM ...`:
@@ -409,6 +497,7 @@ function buildIncludeChildRowsSelect(
       hiddenOrderProjection,
       aggregateOrderBy,
       whereExpr,
+      junctionJoins,
     });
   }
 
@@ -467,6 +556,10 @@ function buildIncludeChildRowsSelect(
     childRows = childRows.withJoins([...polyJoinsAndProjection.joins]);
   }
 
+  if (junctionJoins.length > 0) {
+    childRows = childRows.withJoins(junctionJoins);
+  }
+
   if (childState.distinctOn && childState.distinctOn.length > 0) {
     childRows = childRows.withDistinctOn(
       childState.distinctOn.map((column) => ColumnRef.of(childTableRef, column)),
@@ -529,6 +622,7 @@ function buildDistinctNonLeafChildRowsSelect(options: {
   readonly hiddenOrderProjection: ReadonlyArray<ProjectionItem>;
   readonly aggregateOrderBy: ReadonlyArray<OrderByItem> | undefined;
   readonly whereExpr: AnyExpression;
+  readonly junctionJoins: ReadonlyArray<JoinAst>;
 }): {
   readonly childRows: SelectAst;
   readonly childProjection: ReadonlyArray<ProjectionItem>;
@@ -545,6 +639,7 @@ function buildDistinctNonLeafChildRowsSelect(options: {
     hiddenOrderProjection,
     aggregateOrderBy,
     whereExpr,
+    junctionJoins,
   } = options;
   const childState = include.nested;
 
@@ -614,8 +709,9 @@ function buildDistinctNonLeafChildRowsSelect(options: {
       ...hiddenOrderProjection,
     ])
     .withWhere(whereExpr);
-  if (polyJoinsAndProjection.joins.length > 0) {
-    baseInner = baseInner.withJoins([...polyJoinsAndProjection.joins]);
+  const distinctExtraJoins = [...polyJoinsAndProjection.joins, ...junctionJoins];
+  if (distinctExtraJoins.length > 0) {
+    baseInner = baseInner.withJoins(distinctExtraJoins);
   }
 
   // `childState.distinct` is non-empty by the `isDistinctNonLeaf` guard

packages/3-extensions/sql-orm-client/src/types.ts

@@ -50,6 +50,20 @@ export interface IncludeCombine<ResultShape extends Record<string, unknown>>
   readonly branches: Readonly<Record<string, IncludeCombineBranch>>;
 }
 
+export interface IncludeThroughDescriptor {
+  readonly table: string;
+  /** Namespace the junction table lives in, as declared in the contract. */
+  readonly namespaceId: string;
+  /** FK columns in the junction table that point to the parent. */
+  readonly parentColumns: readonly string[];
+  /** FK columns in the junction table that point to the target (child). */
+  readonly childColumns: readonly string[];
+  /** PK columns in the target table that the junction's childColumns reference. */
+  readonly targetColumns: readonly string[];
+  /** Resolved column names in the parent table that junction.parentColumns reference. */
+  readonly parentLocalColumns: readonly string[];
+}
+
 export interface IncludeExpr {
   readonly relationName: string;
   readonly relatedModelName: string;
@@ -58,6 +72,7 @@ export interface IncludeExpr {
   readonly targetColumn: string;
   readonly localColumn: string;
   readonly cardinality: RelationCardinalityTag | undefined;
+  readonly through?: IncludeThroughDescriptor;
   readonly nested: CollectionState;
   readonly scalar: IncludeScalar<unknown> | undefined;
   readonly combine: Readonly<Record<string, IncludeCombineBranch>> | undefined;