fix(db): align join keys with equality identity

.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.

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<unknown>) => 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<string, LazyCollectionCallbacks>,
   target: { sourceId: string; path: Array<string>; 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]
+    }),
+  )
 }
 
 /**