fix: correlated subquery resolves empty when correlation targets a joined alias

.changeset/plenty-hoops-shake.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/db': patch
+---
+
+Fix: `materialize()` correlated subquery silently resolves to an empty array when the correlation predicate references a joined alias instead of the subquery's own `from` alias. The parent-key filter is now deferred until after joins when the correlation field lives on a joined source.

packages/db/src/query/compiler/index.ts

@@ -243,7 +243,20 @@ export function compileQuery(
   // The inner join happens BEFORE namespace wrapping / WHERE / SELECT / ORDER BY,
   // so the child pipeline only processes rows that match parents.
   let pipeline: NamespacedAndKeyedStream = initialPipeline
-  if (!isUnionFrom && parentKeyStream && childCorrelationField) {
+  // When the correlation field references a joined alias (not the main from
+  // alias), the correlation value isn't available on the raw input rows, so
+  // the parent-key filter must be deferred until after joins are processed.
+  const correlationOnJoinedAlias: boolean =
+    !isUnionFrom &&
+    !!parentKeyStream &&
+    !!childCorrelationField &&
+    childCorrelationField.path[0] !== mainSource
+  if (
+    !isUnionFrom &&
+    parentKeyStream &&
+    childCorrelationField &&
+    !correlationOnJoinedAlias
+  ) {
     const mainInput = sources[mainSource]!
     let filteredMainInput = mainInput
     // Re-key child input by correlation field: [correlationValue, [childKey, childRow]]
@@ -310,6 +323,56 @@ export function compileQuery(
     )
   }
 
+  // Deferred parent-key filter for correlation fields on a joined alias.
+  // Mirrors the pre-join filter above, but operates on namespaced rows so
+  // the joined alias carrying the correlation value is present.
+  if (correlationOnJoinedAlias) {
+    const compiledChildCorrelation = compileExpression(childCorrelationField!)
+    // Re-key by correlation value: [correlationValue, [rowKey, namespacedRow]]
+    const rekeyed = pipeline.pipe(
+      map(([key, namespacedRow]: [unknown, any]) => {
+        const correlationValue = compiledChildCorrelation(namespacedRow)
+        return [correlationValue, [key, namespacedRow]] as [
+          unknown,
+          [unknown, any],
+        ]
+      }),
+    )
+
+    // Inner join: only rows whose correlation key exists in parent keys pass
+    const joinedWithParents = rekeyed.pipe(
+      joinOperator(parentKeyStream!, `inner`),
+    )
+
+    // Extract back to [rowKey, namespacedRow], tagging __correlationKey on the
+    // main source (where downstream routing expects it) and merging any
+    // projected parent context into the namespaced row for WHERE resolution.
+    pipeline = joinedWithParents.pipe(
+      filter(([_correlationValue, [childSide]]: any) => {
+        return childSide != null
+      }),
+      map(([correlationValue, [childSide, parentSide]]: any) => {
+        const [rowKey, namespacedRow] = childSide
+        const tagged: any = {
+          ...namespacedRow,
+          [mainSource]: {
+            ...namespacedRow[mainSource],
+            __correlationKey: correlationValue,
+          },
+        }
+        if (parentSide != null) {
+          Object.assign(tagged, parentSide)
+          tagged.__parentContext = parentSide
+        }
+        const effectiveKey =
+          parentSide != null
+            ? `${String(rowKey)}::${JSON.stringify(parentSide)}`
+            : rowKey
+        return [effectiveKey, tagged]
+      }),
+    )
+  }
+
   // Process the WHERE clause if it exists
   if (query.where && query.where.length > 0) {
     // Apply each WHERE condition as a filter (they are ANDed together)

packages/db/tests/query/includes.test.ts

@@ -6758,5 +6758,106 @@ describe(`includes subqueries`, () => {
         },
       ])
     })
+
+    it(`materialize with correlation on a joined alias returns matching rows`, async () => {
+      // Regression test for #1704: the correlation predicate references the
+      // joined alias (`i`), not the subquery's own from alias (`c`).
+      const collection = createLiveQueryCollection((q) =>
+        q.from({ p: projects }).select(({ p }) => ({
+          id: p.id,
+          name: p.name,
+          comments: materialize(
+            q
+              .from({ c: comments })
+              .innerJoin({ i: issues }, ({ c, i }) => eq(c.issueId, i.id))
+              .where(({ i }) => eq(i.projectId, p.id))
+              .select(({ c }) => ({ id: c.id, body: c.body })),
+          ),
+        })),
+      )
+      await collection.preload()
+
+      const alpha = collection.get(1) as any
+      expect(sortedPlainRows(alpha.comments)).toEqual([
+        { id: 100, body: `Looks bad` },
+        { id: 101, body: `Fixed it` },
+      ])
+
+      const beta = collection.get(2) as any
+      expect(sortedPlainRows(beta.comments)).toEqual([
+        { id: 200, body: `Same bug` },
+      ])
+
+      const gamma = collection.get(3) as any
+      expect(plainRows(gamma.comments)).toEqual([])
+
+      // Incremental updates route through the joined-alias correlation too
+      comments.utils.begin()
+      comments.utils.write({
+        type: `insert`,
+        value: { id: 102, issueId: 11, body: `New comment` },
+      })
+      comments.utils.commit()
+
+      expect(sortedPlainRows((collection.get(1) as any).comments)).toEqual([
+        { id: 100, body: `Looks bad` },
+        { id: 101, body: `Fixed it` },
+        { id: 102, body: `New comment` },
+      ])
+
+      comments.utils.begin()
+      comments.utils.write({
+        type: `delete`,
+        value: { id: 100, issueId: 10, body: `Looks bad` },
+      })
+      comments.utils.commit()
+
+      expect(sortedPlainRows((collection.get(1) as any).comments)).toEqual([
+        { id: 101, body: `Fixed it` },
+        { id: 102, body: `New comment` },
+      ])
+    })
+
+    it(`materialize with joined-alias correlation and spread select of the from alias`, async () => {
+      // Mirrors the exact shape from #1704: `.select(({ from }) => ({ ...from }))`
+      const collection = createLiveQueryCollection((q) =>
+        q.from({ p: projects }).select(({ p }) => ({
+          id: p.id,
+          comments: materialize(
+            q
+              .from({ c: comments })
+              .innerJoin({ i: issues }, ({ c, i }) => eq(c.issueId, i.id))
+              .where(({ i }) => eq(i.projectId, p.id))
+              .select(({ c }) => ({ ...c })),
+          ),
+        })),
+      )
+      await collection.preload()
+
+      const alpha = collection.get(1) as any
+      expect(sortedPlainRows(alpha.comments)).toEqual([
+        { id: 100, issueId: 10, body: `Looks bad` },
+        { id: 101, issueId: 10, body: `Fixed it` },
+      ])
+
+      // Correlation on the from alias keeps working when a join is present
+      const control = createLiveQueryCollection((q) =>
+        q.from({ iss: issues }).select(({ iss }) => ({
+          id: iss.id,
+          comments: materialize(
+            q
+              .from({ c: comments })
+              .innerJoin({ i2: issues }, ({ c, i2 }) => eq(c.issueId, i2.id))
+              .where(({ c }) => eq(c.issueId, iss.id))
+              .select(({ c }) => ({ id: c.id })),
+          ),
+        })),
+      )
+      await control.preload()
+      expect(sortedPlainRows((control.get(10) as any).comments)).toEqual([
+        { id: 100 },
+        { id: 101 },
+      ])
+    })
   })
 })