diff --git a/.changeset/fix-query-join-equality.md b/.changeset/fix-query-join-equality.md new file mode 100644 index 0000000000..384407cf3d --- /dev/null +++ b/.changeset/fix-query-join-equality.md @@ -0,0 +1,5 @@ +--- +'@tanstack/db': patch +--- + +Align live-query join keys with established predicate equality for binary, temporal, Date, and opaque values, including on-demand collection loading, and prevent nullish operands from matching in full and correlated joins. diff --git a/packages/db/src/query/compiler/joins.ts b/packages/db/src/query/compiler/joins.ts index 4dfa935bc8..a1c5ba5c16 100644 --- a/packages/db/src/query/compiler/joins.ts +++ b/packages/db/src/query/compiler/joins.ts @@ -1,5 +1,4 @@ import { - filter, join as joinOperator, map, serializeValue, @@ -16,7 +15,6 @@ import { UnsupportedJoinSourceTypeError, UnsupportedJoinTypeError, } from '../../errors.js' -import { normalizeValue } from '../../utils/comparison.js' import { getParentContextIdentity, getParentContextValue, @@ -69,6 +67,12 @@ export type LazyCollectionCallbacks = { setDemand?: (plan: LazyDemandPlan, keys: Set) => void } +type JoinInputValue = [ + originalKey: unknown, + namespacedRow: NamespacedRow, + joinValue: unknown, +] + let nextLazyDemandPlanId = 0 function parameterizeJoinInputByParentRoutes( @@ -148,6 +152,24 @@ function getRouteJoinKey( ]) } +function getJoinKey( + row: NamespacedRow, + source: string, + side: `main` | `joined`, + value: unknown, + routeJoinedSource: boolean, + valueIdentity: ValueIdentity, +): string { + if (value == null) { + // Serialized equality and route keys are JSON or `~`-prefixed, so these + // side-local sentinels cannot collide with a satisfiable join operand. + return side === `main` ? `\0m` : `\0j` + } + return routeJoinedSource + ? getRouteJoinKey(row, source, value, valueIdentity) + : valueIdentity.serializeEquality(value) +} + export function registerLazyDemandPlan( callbacks: Record, target: { sourceId: string; path: Array; collection: Collection }, @@ -337,15 +359,20 @@ function processJoin( let mainPipeline = pipeline.pipe( map(([currentKey, namespacedRow]) => { // Extract the join key from the main source expression - const value = normalizeValue(compiledMainExpr(namespacedRow)) - const mainKey = routeJoinedSource - ? getRouteJoinKey(namespacedRow, mainSource, value, valueIdentity) - : value - - // Return [joinKey, [originalKey, namespacedRow]] - return [mainKey, [currentKey, namespacedRow]] as [ - unknown, - [string, typeof namespacedRow], + const value = compiledMainExpr(namespacedRow) + const mainKey = getJoinKey( + namespacedRow, + mainSource, + `main`, + value, + routeJoinedSource, + valueIdentity, + ) + + // Keep the raw value for lazy demand; the equality key is graph-local. + return [mainKey, [currentKey, namespacedRow, value]] as [ + string, + JoinInputValue, ] }), ) @@ -357,15 +384,20 @@ function processJoin( const namespacedRow = wrapJoinedInputRow(joinedSource, row) // Extract the join key from the joined source expression - const value = normalizeValue(compiledJoinedExpr(namespacedRow)) - const joinedKey = routeJoinedSource - ? getRouteJoinKey(namespacedRow, joinedSource, value, valueIdentity) - : value - - // Return [joinKey, [originalKey, namespacedRow]] - return [joinedKey, [currentKey, namespacedRow]] as [ - unknown, - [string, typeof namespacedRow], + const value = compiledJoinedExpr(namespacedRow) + const joinedKey = getJoinKey( + namespacedRow, + joinedSource, + `joined`, + value, + routeJoinedSource, + valueIdentity, + ) + + // Keep the raw value for lazy demand; the equality key is graph-local. + return [joinedKey, [currentKey, namespacedRow, value]] as [ + string, + JoinInputValue, ] }), ) @@ -431,18 +463,20 @@ function processJoin( // Set up lazy loading: intercept active side's stream and dynamically load // matching rows from lazy side based on join keys. const activePipelineWithLoading: IStreamBuilder< - [key: unknown, [originalKey: string, namespacedRow: NamespacedRow]] + [key: string, value: JoinInputValue] > = activePipeline.pipe( tap((data) => { - for (const [[joinKey], weight] of data.getInner()) { - if (joinKey == null) continue - const encoded = valueIdentity.serializeEquality(joinKey) - const previous = demandWeights.get(encoded) + for (const [[joinKey, [, , joinValue]], weight] of data.getInner()) { + if (joinValue == null) continue + const previous = demandWeights.get(joinKey) const nextWeight = (previous?.weight ?? 0) + weight if (nextWeight === 0) { - demandWeights.delete(encoded) + demandWeights.delete(joinKey) } else { - demandWeights.set(encoded, { key: joinKey, weight: nextWeight }) + demandWeights.set(joinKey, { + key: previous?.key ?? joinValue, + weight: nextWeight, + }) } } @@ -468,7 +502,7 @@ function processJoin( return mainPipeline.pipe( joinOperator(joinedPipeline, joinClause.type as JoinType), - processJoinResults(joinClause.type), + processJoinResults, ) } @@ -705,71 +739,38 @@ function getFirstFromAlias(query: QueryIR): string | undefined { return getFromSources(query.from)[0]?.alias } -/** - * Processes the results of a join operation - */ -function processJoinResults(joinType: string) { - return function ( - pipeline: IStreamBuilder< - [ - key: string, - [ - [string, NamespacedRow] | undefined, - [string, NamespacedRow] | undefined, - ], - ] - >, - ): NamespacedAndKeyedStream { - return pipeline.pipe( - // Process the join result and handle nulls - filter((result) => { - const [_key, [main, joined]] = result - const mainNamespacedRow = main?.[1] - const joinedNamespacedRow = joined?.[1] - - // Handle different join types - if (joinType === `inner`) { - return !!(mainNamespacedRow && joinedNamespacedRow) - } - - if (joinType === `left`) { - return !!mainNamespacedRow - } - - if (joinType === `right`) { - return !!joinedNamespacedRow - } - - // For full joins, always include - return true - }), - map((result) => { - const [_key, [main, joined]] = result - const mainKey = main?.[0] - const mainNamespacedRow = main?.[1] - const joinedKey = joined?.[0] - const joinedNamespacedRow = joined?.[1] - - // Merge the namespaced rows - const mergedNamespacedRow: NamespacedRow = {} - - // Add main row data if it exists - if (mainNamespacedRow) { - Object.assign(mergedNamespacedRow, mainNamespacedRow) - } +function processJoinResults( + pipeline: IStreamBuilder< + [key: string, [JoinInputValue | undefined, JoinInputValue | undefined]] + >, +): NamespacedAndKeyedStream { + return pipeline.pipe( + map((result) => { + const [_key, [main, joined]] = result + const mainKey = main?.[0] + const mainNamespacedRow = main?.[1] + const joinedKey = joined?.[0] + const joinedNamespacedRow = joined?.[1] + + // Merge the namespaced rows + const mergedNamespacedRow: NamespacedRow = {} + + // Add main row data if it exists + if (mainNamespacedRow) { + Object.assign(mergedNamespacedRow, mainNamespacedRow) + } - // Add joined row data if it exists - if (joinedNamespacedRow) { - Object.assign(mergedNamespacedRow, joinedNamespacedRow) - } + // Add joined row data if it exists + if (joinedNamespacedRow) { + Object.assign(mergedNamespacedRow, joinedNamespacedRow) + } - // We create a composite key that combines the main and joined keys - const resultKey = `[${mainKey},${joinedKey}]` + // We create a composite key that combines the main and joined keys + const resultKey = `[${mainKey},${joinedKey}]` - return [resultKey, mergedNamespacedRow] as [string, NamespacedRow] - }), - ) - } + return [resultKey, mergedNamespacedRow] as [string, NamespacedRow] + }), + ) } /** diff --git a/packages/db/tests/query/join.test.ts b/packages/db/tests/query/join.test.ts index 9917dca4cd..caa7124d47 100644 --- a/packages/db/tests/query/join.test.ts +++ b/packages/db/tests/query/join.test.ts @@ -13,6 +13,7 @@ import { or, } from '../../src/query/index.js' import { createCollection } from '../../src/collection/index.js' +import { createFilterFunctionFromExpression } from '../../src/collection/change-events.js' import { flushPromises, mockSyncCollectionOptions, @@ -49,6 +50,92 @@ const sampleDepartments: Array = [ { id: 3, name: `Marketing`, budget: 60000 }, ] +const equalityJoinCases = [ + { + label: `scalar strings`, + createValues: () => ({ left: `match`, right: `match`, other: `other` }), + }, + { + label: `small binary values`, + createValues: () => ({ + left: new Uint8Array(16).fill(7), + right: new Uint8Array(16).fill(7), + other: new Uint8Array(16).fill(8), + }), + }, + { + label: `large binary values`, + createValues: () => ({ + left: new Uint8Array(200).fill(7), + right: new Uint8Array(200).fill(7), + other: new Uint8Array(200).fill(8), + }), + }, + { + label: `Uint8Array and Buffer values`, + createValues: () => ({ + left: new Uint8Array(16).fill(7), + right: Buffer.alloc(16, 7), + other: new Uint8Array(16).fill(8), + }), + }, + { + label: `Date timestamps`, + createValues: () => ({ + left: new Date(1), + right: new Date(1), + other: new Date(2), + }), + }, + { + label: `Temporal values`, + createValues: () => ({ + left: Temporal.PlainDate.from(`2024-04-05`), + right: Temporal.PlainDate.from(`2024-04-05`), + other: Temporal.PlainDate.from(`2024-04-06`), + }), + }, + { + label: `BigInt values`, + createValues: () => ({ + left: 9007199254740993n, + right: 9007199254740993n, + other: 1n, + }), + }, + { + label: `NaN values`, + createValues: () => ({ left: Number.NaN, right: Number.NaN, other: 0 }), + }, + { + label: `non-finite numbers`, + createValues: () => ({ + left: Number.POSITIVE_INFINITY, + right: Number.POSITIVE_INFINITY, + other: Number.NEGATIVE_INFINITY, + }), + }, + { + label: `opaque shared references`, + createValues: () => { + const shared = { code: 1 } + return { left: shared, right: shared, other: { code: 1 } } + }, + }, +] + +type JoinPair = readonly [number | undefined, number | undefined] + +function sortJoinPairs(pairs: Array): Array { + return pairs.sort( + ([leftA, rightA], [leftB, rightB]) => + (leftA ?? Number.POSITIVE_INFINITY) - + (leftB ?? Number.POSITIVE_INFINITY) || + (rightA ?? Number.POSITIVE_INFINITY) - + (rightB ?? Number.POSITIVE_INFINITY), + ) +} + function createUsersCollection(autoIndex: `off` | `eager` = `eager`) { return createCollection( mockSyncCollectionOptions({ @@ -1047,6 +1134,287 @@ function createJoinTests(autoIndex: `off` | `eager`): void { }) }) + test.each(equalityJoinCases)( + `$label joins agree with equality predicates`, + ({ label, createValues }) => { + type EqualityRow = { id: number; value: unknown } + const { left: leftValue, right: rightValue, other } = createValues() + const leftCollection = createCollection( + mockSyncCollectionOptions({ + id: `equality-left-${autoIndex}-${label}`, + getKey: (row) => row.id, + initialData: [{ id: 1, value: leftValue }], + autoIndex, + }), + ) + const rightCollection = createCollection( + mockSyncCollectionOptions({ + id: `equality-right-${autoIndex}-${label}`, + getKey: (row) => row.id, + initialData: [ + { id: 10, value: rightValue }, + { id: 20, value: other }, + ], + autoIndex, + }), + ) + const joined = createLiveQueryCollection({ + startSync: true, + query: (q) => + q + .from({ left: leftCollection }) + .innerJoin({ right: rightCollection }, ({ left, right }) => + eq(left.value, right.value), + ) + .select(({ left, right }) => ({ + leftId: left.id, + rightId: right.id, + })), + }) + const filtered = createLiveQueryCollection({ + startSync: true, + query: (q) => + q + .from({ right: rightCollection }) + .where(({ right: row }) => eq(row.value, leftValue)) + .select(({ right: row }) => ({ rightId: row.id })), + }) + + expect(joined.toArray.map(stripVirtualProps)).toEqual([ + { leftId: 1, rightId: 10 }, + ]) + expect(filtered.toArray.map(stripVirtualProps)).toEqual([ + { rightId: 10 }, + ]) + }, + ) + + test(`binary values stay disjoint from normalization-like strings`, () => { + type EqualityRow = { id: number; value: unknown } + const bytes = new Uint8Array([65]) + const text = `\u0000tanstack-db:binary:A` + const leftCollection = createCollection( + mockSyncCollectionOptions({ + id: `binary-disjoint-left-${autoIndex}`, + getKey: (row) => row.id, + initialData: [{ id: 1, value: bytes }], + autoIndex, + }), + ) + const rightCollection = createCollection( + mockSyncCollectionOptions({ + id: `binary-disjoint-right-${autoIndex}`, + getKey: (row) => row.id, + initialData: [ + { id: 10, value: new Uint8Array(bytes) }, + { id: 20, value: text }, + ], + autoIndex, + }), + ) + const query = createLiveQueryCollection({ + startSync: true, + query: (q) => + q + .from({ left: leftCollection }) + .innerJoin({ right: rightCollection }, ({ left, right }) => + eq(left.value, right.value), + ) + .select(({ right }) => ({ rightId: right.id })), + }) + + expect(query.toArray.map(stripVirtualProps)).toEqual([{ rightId: 10 }]) + }) + + test(`full joins leave nullish equality operands unmatched`, () => { + type NullishRow = { id: number; value: null | undefined | string } + const leftCollection = createCollection( + mockSyncCollectionOptions({ + id: `nullish-full-left-${autoIndex}`, + getKey: (row) => row.id, + initialData: [ + { id: 1, value: null }, + { id: 2, value: undefined }, + { id: 3, value: `\0m` }, + { id: 4, value: `\0j` }, + ], + autoIndex, + }), + ) + const rightCollection = createCollection( + mockSyncCollectionOptions({ + id: `nullish-full-right-${autoIndex}`, + getKey: (row) => row.id, + initialData: [ + { id: 1, value: null }, + { id: 2, value: undefined }, + { id: 3, value: `\0m` }, + { id: 4, value: `\0j` }, + ], + autoIndex, + }), + ) + const query = createLiveQueryCollection({ + startSync: true, + query: (q) => + q + .from({ left: leftCollection }) + .fullJoin({ right: rightCollection }, ({ left, right }) => + eq(left.value, right.value), + ) + .select(({ left, right }) => ({ + leftId: left.id, + rightId: right.id, + })), + }) + + const pairs = sortJoinPairs( + query.toArray.map( + ({ leftId, rightId }) => [leftId, rightId] as const, + ), + ) + expect(pairs).toEqual([ + [1, undefined], + [2, undefined], + [3, 3], + [4, 4], + [undefined, 1], + [undefined, 2], + ]) + }) + + test(`binary join identity survives equal and unequal replacements`, () => { + type BinaryRow = { id: number; value: Uint8Array } + const leftCollection = createCollection( + mockSyncCollectionOptions({ + id: `binary-lifecycle-left-${autoIndex}`, + getKey: (row) => row.id, + initialData: [{ id: 1, value: new Uint8Array([1, 2, 3]) }], + autoIndex, + }), + ) + const rightCollection = createCollection( + mockSyncCollectionOptions({ + id: `binary-lifecycle-right-${autoIndex}`, + getKey: (row) => row.id, + initialData: [{ id: 10, value: new Uint8Array([1, 2, 3]) }], + autoIndex, + }), + ) + const query = createLiveQueryCollection({ + startSync: true, + query: (q) => + q + .from({ left: leftCollection }) + .fullJoin({ right: rightCollection }, ({ left, right }) => + eq(left.value, right.value), + ) + .select(({ left, right }) => ({ + leftId: left.id, + rightId: right.id, + })), + }) + const pairs = () => + sortJoinPairs( + query.toArray.map( + ({ leftId, rightId }) => [leftId, rightId] as const, + ), + ) + const replaceRight = (value: Uint8Array) => { + rightCollection.utils.begin() + rightCollection.utils.write({ + type: `update`, + value: { id: 10, value }, + }) + rightCollection.utils.commit() + } + + expect(pairs()).toEqual([[1, 10]]) + replaceRight(new Uint8Array([1, 2, 3])) + expect(pairs()).toEqual([[1, 10]]) + replaceRight(new Uint8Array([1, 2, 4])) + expect(pairs()).toEqual([ + [1, undefined], + [undefined, 10], + ]) + replaceRight(new Uint8Array([1, 2, 3])) + expect(pairs()).toEqual([[1, 10]]) + }) + + test(`nullish outer rows transition through a finite join key`, () => { + type NullableRow = { id: number; value: number | null } + const leftCollection = createCollection( + mockSyncCollectionOptions({ + id: `nullish-lifecycle-left-${autoIndex}`, + getKey: (row) => row.id, + initialData: [{ id: 1, value: null }], + autoIndex, + }), + ) + const rightCollection = createCollection( + mockSyncCollectionOptions({ + id: `nullish-lifecycle-right-${autoIndex}`, + getKey: (row) => row.id, + initialData: [ + { id: 2, value: null }, + { id: 1, value: null }, + ], + autoIndex, + }), + ) + const query = createLiveQueryCollection({ + startSync: true, + query: (q) => + q + .from({ left: leftCollection }) + .fullJoin({ right: rightCollection }, ({ left, right }) => + eq(left.value, right.value), + ) + .select(({ left, right }) => ({ + leftId: left.id, + rightId: right.id, + })), + }) + const pairs = () => + sortJoinPairs( + query.toArray + .map(({ leftId, rightId }) => [leftId, rightId] as const) + .reverse(), + ) + const update = ( + collection: typeof leftCollection, + id: number, + value: number | null, + ) => { + collection.utils.begin() + collection.utils.write({ type: `update`, value: { id, value } }) + collection.utils.commit() + } + + expect(pairs()).toEqual([ + [1, undefined], + [undefined, 1], + [undefined, 2], + ]) + update(rightCollection, 1, 1) + expect(pairs()).toEqual([ + [1, undefined], + [undefined, 1], + [undefined, 2], + ]) + update(leftCollection, 1, 1) + expect(pairs()).toEqual([ + [1, 1], + [undefined, 2], + ]) + update(leftCollection, 1, null) + expect(pairs()).toEqual([ + [1, undefined], + [undefined, 1], + [undefined, 2], + ]) + }) + test(`should update Date join matches when timestamp changes`, () => { type DateLeft = { id: number; joinedAt: Date; name: string } type DateRight = { id: number; joinedAt: Date; label: string } @@ -2115,3 +2483,99 @@ describe(`Query JOIN Operations`, () => { createJoinTests(`off`) createJoinTests(`eager`) }) + +test.each([`off`, `eager`] as const)( + `lazy binary join demand uses predicate equality with autoIndex %s`, + async (autoIndex) => { + type BinaryRow = { id: number; binaryId: Uint8Array } + const activeKey = new Uint8Array([1, 2, 3]) + const active = createCollection( + mockSyncCollectionOptions({ + id: `binary-demand-active-${autoIndex}`, + getKey: (row) => row.id, + initialData: [{ id: 1, binaryId: activeKey }], + autoIndex, + }), + ) + const backend: Array = [ + { id: 10, binaryId: new Uint8Array(activeKey) }, + { id: 20, binaryId: new Uint8Array([1, 2, 4]) }, + ] + let loadCalls = 0 + let candidateChecks = 0 + const requestedValues: Array = [] + const lazy = createCollection( + mockSyncCollectionOptions({ + id: `binary-demand-lazy-${autoIndex}`, + getKey: (row) => row.id, + initialData: [], + autoIndex, + syncMode: `on-demand`, + sync: { + sync: (actions) => { + actions.markReady() + return { + loadSubset: (options) => { + loadCalls++ + expect(options.where).toBeDefined() + const where = options.where! + expect(where.type).toBe(`func`) + if (where.type !== `func` || where.name !== `in`) { + throw new Error(`expected binary demand to use IN`) + } + const candidates = where.args[1] + if ( + candidates?.type !== `val` || + !Array.isArray(candidates.value) + ) { + throw new Error(`expected binary demand candidates`) + } + requestedValues.push(...candidates.value) + const matches = + createFilterFunctionFromExpression(where) + actions.begin() + for (const row of backend) { + candidateChecks++ + if (matches(row)) { + actions.write({ type: `insert`, value: row }) + } + } + const applied = actions.commit() + return applied === true ? true : applied + }, + unloadSubset: () => {}, + } + }, + }, + }), + ) + const query = createLiveQueryCollection({ + query: (q) => + q + .from({ left: active }) + .leftJoin({ right: lazy }, ({ left, right }) => + eq(left.binaryId, right.binaryId), + ) + .select(({ left, right }) => ({ + leftId: left.id, + rightId: right.id, + })), + }) + + try { + await query.preload() + expect(loadCalls).toBe(1) + expect(candidateChecks).toBe(2) + expect(requestedValues).toHaveLength(1) + expect(requestedValues[0]).toBeInstanceOf(Uint8Array) + expect(Array.from(requestedValues[0] as Uint8Array)).toEqual( + Array.from(activeKey), + ) + const rows = query.toArray.map(stripVirtualProps) + const expected = [{ leftId: 1, rightId: 10 }] + expect(rows).toEqual(expected) + } finally { + await Promise.all([query.cleanup(), lazy.cleanup(), active.cleanup()]) + } + }, +)