Skip to content

Commit 1814f8c

Browse files
kevin-dpkevin-dp
authored
Deoptimize join when lazy side has a limit/offset clause (#508)
1 parent 0be4e2c commit 1814f8c

3 files changed

Lines changed: 178 additions & 79 deletions

File tree

‎.changeset/modern-trees-mate.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
"@tanstack/db": patch
3+
---
4+
5+
Fix bug where too much data would be loaded when the lazy collection of a join contains an offset and/or limit clause.

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

Lines changed: 95 additions & 79 deletions
Original file line numberDiff line numberDiff line change
@@ -187,90 +187,106 @@ function processJoin(
187187
}
188188

189189
if (activeCollection) {
190-
// This join can be optimized by having the active collection
191-
// dynamically load keys into the lazy collection
192-
// based on the value of the joinKey and by looking up
193-
// matching rows in the index of the lazy collection
194-
195-
// Mark the lazy collection as lazy
196-
// this Set is passed by the liveQueryCollection to the compiler
197-
// such that the liveQueryCollection can check it after compilation
198-
// to know which collections are lazy collections
199-
lazyCollections.add(lazyCollection.id)
200-
201-
const activePipeline =
202-
activeCollection === `main` ? mainPipeline : joinedPipeline
203-
204-
let index: BaseIndex<string | number> | undefined
205-
206-
const lazyCollectionJoinExpr =
207-
activeCollection === `main`
208-
? (joinedExpr as PropRef)
209-
: (mainExpr as PropRef)
210-
211-
const followRefResult = followRef(
212-
rawQuery,
213-
lazyCollectionJoinExpr,
214-
lazyCollection
215-
)!
216-
const followRefCollection = followRefResult.collection
217-
218-
const fieldName = followRefResult.path[0]
219-
if (fieldName) {
220-
ensureIndexForField(fieldName, followRefResult.path, followRefCollection)
221-
}
222-
223-
let deoptimized = false
224-
225-
const activePipelineWithLoading: IStreamBuilder<
226-
[key: unknown, [originalKey: string, namespacedRow: NamespacedRow]]
227-
> = activePipeline.pipe(
228-
tap(([joinKey, _]) => {
229-
if (deoptimized) {
230-
return
231-
}
232-
233-
// Find the index for the path we join on
234-
// we need to find the index inside the map operator
235-
// because the indexes are only available after the initial sync
236-
// so we can't fetch it during compilation
237-
index ??= findIndexForField(
238-
followRefCollection.indexes,
239-
followRefResult.path
190+
// If the lazy collection comes from a subquery that has a limit and/or an offset clause
191+
// then we need to deoptimize the join because we don't know which rows are in the result set
192+
// since we simply lookup matching keys in the index but the index contains all rows
193+
// (not just the ones that pass the limit and offset clauses)
194+
const lazyFrom =
195+
activeCollection === `main` ? joinClause.from : rawQuery.from
196+
const limitedSubquery =
197+
lazyFrom.type === `queryRef` &&
198+
(lazyFrom.query.limit || lazyFrom.query.offset)
199+
200+
if (!limitedSubquery) {
201+
// This join can be optimized by having the active collection
202+
// dynamically load keys into the lazy collection
203+
// based on the value of the joinKey and by looking up
204+
// matching rows in the index of the lazy collection
205+
206+
// Mark the lazy collection as lazy
207+
// this Set is passed by the liveQueryCollection to the compiler
208+
// such that the liveQueryCollection can check it after compilation
209+
// to know which collections are lazy collections
210+
lazyCollections.add(lazyCollection.id)
211+
212+
const activePipeline =
213+
activeCollection === `main` ? mainPipeline : joinedPipeline
214+
215+
let index: BaseIndex<string | number> | undefined
216+
217+
const lazyCollectionJoinExpr =
218+
activeCollection === `main`
219+
? (joinedExpr as PropRef)
220+
: (mainExpr as PropRef)
221+
222+
const followRefResult = followRef(
223+
rawQuery,
224+
lazyCollectionJoinExpr,
225+
lazyCollection
226+
)!
227+
const followRefCollection = followRefResult.collection
228+
229+
const fieldName = followRefResult.path[0]
230+
if (fieldName) {
231+
ensureIndexForField(
232+
fieldName,
233+
followRefResult.path,
234+
followRefCollection
240235
)
236+
}
241237

242-
// The `callbacks` object is passed by the liveQueryCollection to the compiler.
243-
// It contains a function to lazy load keys for each lazy collection
244-
// as well as a function to switch back to a regular collection
245-
// (useful when there's no index for available for lazily loading the collection)
246-
const collectionCallbacks = callbacks[lazyCollection.id]
247-
if (!collectionCallbacks) {
248-
throw new Error(
249-
`Internal error: callbacks for collection are missing in join pipeline. Make sure the live query collection sets them before running the pipeline.`
238+
let deoptimized = false
239+
240+
const activePipelineWithLoading: IStreamBuilder<
241+
[key: unknown, [originalKey: string, namespacedRow: NamespacedRow]]
242+
> = activePipeline.pipe(
243+
tap(([joinKey, _]) => {
244+
if (deoptimized) {
245+
return
246+
}
247+
248+
// Find the index for the path we join on
249+
// we need to find the index inside the map operator
250+
// because the indexes are only available after the initial sync
251+
// so we can't fetch it during compilation
252+
index ??= findIndexForField(
253+
followRefCollection.indexes,
254+
followRefResult.path
250255
)
251-
}
252256

253-
const { loadKeys, loadInitialState } = collectionCallbacks
254-
255-
if (index && index.supports(`eq`)) {
256-
// Use the index to fetch the PKs of the rows in the lazy collection
257-
// that match this row from the active collection based on the value of the joinKey
258-
const matchingKeys = index.lookup(`eq`, joinKey)
259-
// Inform the lazy collection that those keys need to be loaded
260-
loadKeys(matchingKeys)
261-
} else {
262-
// We can't optimize the join because there is no index for the join key
263-
// on the lazy collection, so we load the initial state
264-
deoptimized = true
265-
loadInitialState()
266-
}
267-
})
268-
)
257+
// The `callbacks` object is passed by the liveQueryCollection to the compiler.
258+
// It contains a function to lazy load keys for each lazy collection
259+
// as well as a function to switch back to a regular collection
260+
// (useful when there's no index for available for lazily loading the collection)
261+
const collectionCallbacks = callbacks[lazyCollection.id]
262+
if (!collectionCallbacks) {
263+
throw new Error(
264+
`Internal error: callbacks for collection are missing in join pipeline. Make sure the live query collection sets them before running the pipeline.`
265+
)
266+
}
267+
268+
const { loadKeys, loadInitialState } = collectionCallbacks
269+
270+
if (index && index.supports(`eq`)) {
271+
// Use the index to fetch the PKs of the rows in the lazy collection
272+
// that match this row from the active collection based on the value of the joinKey
273+
const matchingKeys = index.lookup(`eq`, joinKey)
274+
// Inform the lazy collection that those keys need to be loaded
275+
loadKeys(matchingKeys)
276+
} else {
277+
// We can't optimize the join because there is no index for the join key
278+
// on the lazy collection, so we load the initial state
279+
deoptimized = true
280+
loadInitialState()
281+
}
282+
})
283+
)
269284

270-
if (activeCollection === `main`) {
271-
mainPipeline = activePipelineWithLoading
272-
} else {
273-
joinedPipeline = activePipelineWithLoading
285+
if (activeCollection === `main`) {
286+
mainPipeline = activePipelineWithLoading
287+
} else {
288+
joinedPipeline = activePipelineWithLoading
289+
}
274290
}
275291
}
276292

‎packages/db/tests/query/join-subquery.test.ts‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -432,6 +432,84 @@ function createJoinSubqueryTests(autoIndex: `off` | `eager`): void {
432432
})
433433
})
434434

435+
test(`should use subquery in LEFT JOIN clause - left join with ordered subquery with limit`, () => {
436+
const joinSubquery = createLiveQueryCollection({
437+
query: (q) => {
438+
return q
439+
.from({ issue: issuesCollection })
440+
.join(
441+
{
442+
users: q
443+
.from({ user: usersCollection })
444+
.where(({ user }) => eq(user.status, `active`))
445+
.orderBy(({ user }) => user.name, `asc`)
446+
.limit(1),
447+
},
448+
({ issue, users }) => eq(issue.userId, users.id),
449+
`left`
450+
)
451+
.orderBy(({ issue }) => issue.id, `desc`)
452+
.limit(1)
453+
},
454+
startSync: true,
455+
})
456+
457+
const results = joinSubquery.toArray
458+
console.log(`results`, results)
459+
expect(results).toEqual([
460+
{
461+
issue: {
462+
id: 5,
463+
title: `Feature 2`,
464+
status: `in_progress`,
465+
projectId: 2,
466+
userId: 2,
467+
duration: 15,
468+
createdAt: `2024-01-05`,
469+
},
470+
},
471+
])
472+
})
473+
474+
test(`should use subquery in RIGHT JOIN clause - left join with ordered subquery with limit`, () => {
475+
const joinSubquery = createLiveQueryCollection({
476+
query: (q) => {
477+
return q
478+
.from({
479+
users: q
480+
.from({ user: usersCollection })
481+
.where(({ user }) => eq(user.status, `active`))
482+
.orderBy(({ user }) => user.name, `asc`)
483+
.limit(1),
484+
})
485+
.join(
486+
{ issue: issuesCollection },
487+
({ issue, users }) => eq(issue.userId, users.id),
488+
`right`
489+
)
490+
.orderBy(({ issue }) => issue.id, `desc`)
491+
.limit(1)
492+
},
493+
startSync: true,
494+
})
495+
496+
const results = joinSubquery.toArray
497+
console.log(`results`, results)
498+
expect(results).toEqual([
499+
{
500+
issue: {
501+
id: 5,
502+
title: `Feature 2`,
503+
status: `in_progress`,
504+
projectId: 2,
505+
userId: 2,
506+
duration: 15,
507+
createdAt: `2024-01-05`,
508+
},
509+
},
510+
])
511+
})
512+
435513
test(`should handle subqueries with SELECT clauses in both FROM and JOIN`, () => {
436514
const joinQuery = createLiveQueryCollection({
437515
startSync: true,

0 commit comments

Comments
 (0)