fix(db): keep join-key demand when an include filters a join side

.changeset/lazy-join-keeps-key-filter.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/db': patch
+---
+
+Keep loading the lazy side of a join inside an include by its join keys when the query also filters one side with `where`. The filter now narrows each keyed `loadSubset` request (for example `active = true AND id IN (7, 8, 9)`) instead of replacing it, so an on-demand Collection no longer receives a request for every row that matches the filter, or for every row when the filter is on the other side.

docs/contributing/oracle-coverage.md

@@ -1210,6 +1210,30 @@ fixed collapse and the detail page; a child that takes over `setWindow` fails
 the rows law. Focused tests require a window inside a `from` or joined subquery
 not to answer `setWindow`.
 
+Its two joined work laws (unmatched links, and indexed join targets) also
+generate a join filter: none, a `where` on the links alone, or a `where` on the
+joined targets alone. The optimizer pushes each filter into a subquery over its
+own source, and the include must still filter the links by parent and load the
+targets by the kept links' join values, so the filter narrows each keyed read
+(live/ARCHITECTURE.md, Initial demand and Work). Both campaigns declare the
+three filter states with `createOracleReach` and run one fixed example per
+state; each state is reached by at least one fixed history in every run. On
+main the include parent-routed both filtered sources and scanned the target
+Collection: the target filter delivered 3 and examined 4 target rows at
+baseline where the keyed bound is 2 and 3, and one unmatched link with a link
+filter delivered 4 target rows where the bound is 2. Mutants that route the
+filtered target, or the filtered links, by parent fail the target and link
+states respectively. A focused test in
+`packages/db/tests/query/lazy-join-source-where.test.ts` replays the on-demand
+report: each `loadSubset` the lazy side receives names the join keys. The
+owner does not generate subqueries that read a parent value, filters on both
+sides at once, or `fn.where()` filters, which remain parent-routed. Only a
+focused test in the same file shows that a joined subquery whose `where` reads
+the parent stays parent-routed: a mutant that drops that check returns empty
+child arrays there, and the existing include suites (637 tests) pass it. A
+generated parent-reading subquery dimension in the cross-formulation owner
+would close that gap.
+
 Recorded limit (#2151): while an index-backed parent top-K refills within one
 graph run (a held row moved out, a tie group loaded at the window edge, a
 forward window move), it can briefly hold a row outside the window it shows once

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

@@ -66,7 +66,12 @@ import {
   isCaseWhenConditionTrue,
   toBooleanPredicate,
 } from './evaluators.js'
-import { processJoins, processSource, registerLazyDemandPlan } from './joins.js'
+import {
+  isSourceFilter,
+  processJoins,
+  processSource,
+  registerLazyDemandPlan,
+} from './joins.js'
 import { findAggregate } from './find-aggregate.js'
 import { getLazyLoadTargets } from './lazy-targets.js'
 import { processOrderBy } from './order-by.js'
@@ -1652,6 +1657,12 @@ function processFromClause(
   }
 
   const compileSource = (source: CollectionSourceRef | QueryRef) => {
+    // A pushed-down source filter reads no parent value: it is filtered by the
+    // parent like its Collection, so keyed lazy demand still applies.
+    const isParentRouted =
+      parentKeyStream !== undefined &&
+      source.type === `queryRef` &&
+      !isSourceFilter(source)
     const result = processSource(
       source,
       allInputs,
@@ -1678,7 +1689,7 @@ function processFromClause(
       sourceWhereClauses,
       valueIdentity,
       `from`,
-      parentKeyStream,
+      isParentRouted ? parentKeyStream : undefined,
     )
     return {
       ...result,
@@ -1687,8 +1698,7 @@ function processFromClause(
           sourceAlias: source.alias,
           include,
         })) ?? [],
-      isParentRouted:
-        source.type === `queryRef` && parentKeyStream !== undefined,
+      isParentRouted,
     }
   }
 

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

@@ -24,6 +24,7 @@ import {
   collectPropRefs,
   getFromSources,
   getPropRefSourceAlias,
+  getWhereExpression,
   requireCollectionSource,
 } from '../ir.js'
 import { compileExpression } from './evaluators.js'
@@ -250,7 +251,8 @@ export function processJoins(
     )
     const routeJoinedSource =
       parentKeyStream !== undefined &&
-      (joinClause.from.type === `queryRef` ||
+      ((joinClause.from.type === `queryRef` &&
+        !isSourceFilter(joinClause.from)) ||
         ((joinClause.type === `right` || joinClause.type === `full`) &&
           !mainSourceIsParentFiltered) ||
         joinedExpressionUsesParent)
@@ -475,6 +477,23 @@ export function processJoins(
 
 type JoinRefRole = `joined` | `main` | `ancestor` | `unknown`
 
+// The optimizer pushes a source's own `where` into a subquery over that
+// source. The subquery reads no parent value, so an include need not route it
+// by parent: it joins like the source and keeps keyed lazy demand, and the
+// filter narrows each keyed load.
+export function isSourceFilter({ query }: QueryRef): boolean {
+  const { from, where = [], ...rest } = query
+  return (
+    (from.type === `collectionRef` || from.type === `descriptorRef`) &&
+    Object.values<unknown>(rest).every((value) => value === undefined) &&
+    where.every((clause) =>
+      collectPropRefs(getWhereExpression(clause)).every(
+        (ref) => ref.bindingId === from.bindingId,
+      ),
+    )
+  )
+}
+
 function getJoinRefAlias(ref: PropRef): string | undefined {
   return getPropRefSourceAlias(ref) ?? ref.path[0]
 }

packages/db/tests/query/includes-work-counter-oracle.test.ts

@@ -41,6 +41,7 @@ import type { Collection } from '../../src/collection/index.js'
 import type { LoadSubsetOptions } from '../../src/types.js'
 import type { Context, QueryBuilder } from '../../src/query/index.js'
 import type {
+  JoinFilter,
   SourceWork,
   WorkObservation,
 } from './includes-work-counter-support.js'
@@ -66,9 +67,15 @@ import type {
  * The input grammar varies an integer filler count from 1 through 24. Zero is
  * the ablated baseline; negative and fractional counts are invalid fixture
  * sizes. Joined queries vary unrelated links and join targets separately.
- * The join-free query grows every source together. Counts 1, 2, and 3 cover the
- * smallest valid boundary, 24 covers the upper bound, and fixed and random
- * campaigns sample the range.
+ * Both joined laws also vary a join filter: none, a `where` on the links
+ * alone, or a `where` on the joined targets alone. The optimizer pushes each
+ * filter into its own source, and laws 2 and 3 must still hold: the targets
+ * stay keyed by the kept links' join values, so a filter narrows the keyed
+ * load and must not replace it or make the include scan a side
+ * (live/ARCHITECTURE.md, Initial demand and Work). The reference drops the
+ * filtered row's link. The join-free query grows every source together.
+ * Counts 1, 2, and 3 cover the smallest valid boundary, 24 covers the upper
+ * bound, and fixed and random campaigns sample the range.
  *
  * The boundary conditions matter. Collections preload before they receive
  * B-tree indexes. Filler rows never match a selected route. This suite checks
@@ -85,54 +92,72 @@ const joinedBaselineWork: SourceWork = {
   links: { delivered: 2, examined: 2 },
 }
 
+// Each join filter drops one row after a keyed read. A target filter drops
+// one of the two named targets: the root and both named targets are
+// examined, the root and one target delivered. A link filter drops one of
+// the group's two links, so only the kept link's target is named.
+const joinedBaselineWorkByFilter: Record<JoinFilter, SourceWork> = {
+  none: joinedBaselineWork,
+  target: { ...joinedBaselineWork, terms: { delivered: 2, examined: 3 } },
+  link: {
+    ...joinedBaselineWork,
+    terms: { delivered: 2, examined: 2 },
+    links: { delivered: 1, examined: 2 },
+  },
+}
+const joinFilters = Object.keys(joinedBaselineWorkByFilter) as Array<JoinFilter>
+
 const joinFreeBaselineWork: SourceWork = {
   terms: { delivered: 1, examined: 1 },
   meanings: { delivered: 1, examined: 1 },
   groups: { delivered: 1, examined: 1 },
   links: { delivered: 2, examined: 2 },
 }
 
-let joinedBaselineObservation: WorkObservation
+const joinedBaselineObservations = new Map<JoinFilter, WorkObservation>()
 let joinFreeBaselineObservation: WorkObservation
 
-async function expectCorrelatedJoinWorkBound(
-  fillerCount: number,
+// Both joined laws compare a scaled run with the minimal run of the same
+// query, under each join filter.
+async function expectJoinedWorkBound(
+  filler: { terms: number; links: number },
+  joinFilter: JoinFilter,
 ): Promise<void> {
-  const baseline = joinedBaselineObservation
+  const baseline = joinedBaselineObservations.get(joinFilter)!
   const scaled = await observeWork({
-    filler: {
-      terms: 0,
-      meanings: 0,
-      groups: 0,
-      links: fillerCount,
-    },
+    filler: { ...filler, meanings: 0, groups: 0 },
     joinTargets: true,
+    joinFilter,
   })
-  expect(baseline.result).toEqual(expectedResult({ joinTargets: true }))
+  expect(baseline.result).toEqual(
+    expectedResult({ joinTargets: true, joinFilter }),
+  )
   expect(scaled.result).toEqual(baseline.result)
-  expect(baseline.sourceWork).toEqual(joinedBaselineWork)
-
+  expect(baseline.sourceWork).toEqual(joinedBaselineWorkByFilter[joinFilter])
   expectSourceWorkBound(scaled.sourceWork, baseline.sourceWork)
 }
 
-async function expectJoinTargetWorkBound(fillerCount: number): Promise<void> {
-  const baseline = joinedBaselineObservation
-  const scaled = await observeWork({
-    filler: {
-      terms: fillerCount,
-      meanings: 0,
-      groups: 0,
-      links: 0,
-    },
-    joinTargets: true,
-  })
+async function expectCorrelatedJoinWorkBound(
+  fillerCount: number,
+  joinFilter: JoinFilter = `none`,
+): Promise<void> {
+  await expectJoinedWorkBound({ terms: 0, links: fillerCount }, joinFilter)
+}
 
-  expect(baseline.result).toEqual(expectedResult({ joinTargets: true }))
-  expect(scaled.result).toEqual(baseline.result)
-  expect(baseline.sourceWork).toEqual(joinedBaselineWork)
-  expectSourceWorkBound(scaled.sourceWork, baseline.sourceWork)
+async function expectJoinTargetWorkBound(
+  fillerCount: number,
+  joinFilter: JoinFilter = `none`,
+): Promise<void> {
+  await expectJoinedWorkBound({ terms: fillerCount, links: 0 }, joinFilter)
 }
 
+// The joined campaigns declare every join-filter state and fail unless their
+// own generated runs enter each of them.
+const joinedCase: fc.Arbitrary<readonly [number, JoinFilter]> = fc.tuple(
+  fc.integer({ min: 1, max: 24 }),
+  fc.constantFrom(...joinFilters),
+)
+
 async function expectJoinFreeWorkBound(fillerCount: number): Promise<void> {
   const baseline = joinFreeBaselineObservation
   const scaled = await observeWork({
@@ -166,27 +191,36 @@ function campaigns(fixedSeed: number, property: string, baseRuns = 6) {
 
 describe(`includes deterministic work-counter oracle`, () => {
   beforeAll(async () => {
-    const [joinedBaseline, joinFreeBaseline] = await Promise.all([
-      observeWork({ filler: noFillers, joinTargets: true }),
+    const [joinFreeBaseline, ...joinedBaselines] = await Promise.all([
       observeWork({ filler: noFillers, joinTargets: false }),
+      ...joinFilters.map((joinFilter) =>
+        observeWork({ filler: noFillers, joinTargets: true, joinFilter }),
+      ),
     ])
-    joinedBaselineObservation = joinedBaseline
     joinFreeBaselineObservation = joinFreeBaseline
+    joinFilters.forEach((joinFilter, index) =>
+      joinedBaselineObservations.set(joinFilter, joinedBaselines[index]!),
+    )
   })
 
   it.each([1, 2, 3])(
     `checks the work bound at the small filler boundary (%i)`,
-    (...args: Parameters<typeof expectCorrelatedJoinWorkBound>) => {
+    (fillerCount: number) => {
       oracleRoute(`case`)
-      return expectCorrelatedJoinWorkBound(...args)
+      return expectCorrelatedJoinWorkBound(fillerCount)
     },
   )
   it(`checks the joined-link filler range endpoint`, () => {
     oracleRoute(`fixed history`)
     return expectCorrelatedJoinWorkBound(24)
   })
-  it.each([1, 24])(
-    `checks join-target growth at filler count %i`,
+  it.each(
+    joinFilters.flatMap((joinFilter) => [
+      [1, joinFilter] as const,
+      [24, joinFilter] as const,
+    ]),
+  )(
+    `checks join-target growth at filler count %i (join filter %s)`,
     (...args: Parameters<typeof expectJoinTargetWorkBound>) => {
       oracleRoute(`case`)
       return expectJoinTargetWorkBound(...args)
@@ -200,24 +234,37 @@ describe(`includes deterministic work-counter oracle`, () => {
     },
   )
 
-  for (const campaign of campaigns(1709, `includes-work.correlated-links`)) {
-    fcTest.prop([fc.integer({ min: 1, max: 24 })], campaign.options)(
-      `a join preserves correlated source pushdown (${campaign.name})`,
-      (...args: Parameters<typeof expectCorrelatedJoinWorkBound>) => {
-        oracleRoute(oracleRouteOf(campaign.options))
-        return expectCorrelatedJoinWorkBound(...args)
-      },
-    )
-  }
-
-  for (const campaign of campaigns(170_900, `includes-work.join-targets`)) {
-    fcTest.prop([fc.integer({ min: 1, max: 24 })], campaign.options)(
-      `indexed join-target growth keeps source work flat (${campaign.name})`,
-      (...args: Parameters<typeof expectJoinTargetWorkBound>) => {
+  for (const [property, seed, check] of [
+    [`includes-work.correlated-links`, 1709, expectCorrelatedJoinWorkBound],
+    [`includes-work.join-targets`, 170_900, expectJoinTargetWorkBound],
+  ] as const) {
+    for (const campaign of campaigns(seed, property)) {
+      it(`${property} keeps source work flat under each join filter (${campaign.name})`, async () => {
+        // One fixed example per join filter makes reach independent of the seed.
+        const fixed = joinFilters.map(
+          (joinFilter): readonly [number, JoinFilter] => [1, joinFilter],
+        )
+        const reach = createOracleReach(
+          `${property} (${campaign.name})`,
+          joinFilters,
+          { fixedHistories: fixed.length },
+        )
         oracleRoute(oracleRouteOf(campaign.options))
-        return expectJoinTargetWorkBound(...args)
-      },
-    )
+        await fc.assert(
+          fc.asyncProperty(joinedCase, async (testCase) => {
+            const [fillerCount, joinFilter] = testCase
+            reach.beginHistory({ fixed: fixed.includes(testCase) })
+            reach.reach(joinFilter)
+            await check(fillerCount, joinFilter)
+          }),
+          {
+            ...campaign.options,
+            examples: fixed.map((testCase) => [testCase]),
+          },
+        )
+        reach.assertReached()
+      })
+    }
   }
 
   for (const campaign of campaigns(17_090, `includes-work.join-free`)) {
@@ -1091,12 +1138,11 @@ declareOracle({
   ],
   check: [
     checkWindowedIncludeWork,
-    expectCorrelatedJoinWorkBound,
-    expectJoinTargetWorkBound,
+    expectJoinedWorkBound,
     expectJoinFreeWorkBound,
   ],
   model: [windowUnits, modelKids],
-  grammar: [windowCase],
+  grammar: [windowCase, joinedCase],
 })
 
 // Enough runs that each dimension appears at the default multiplier; the