Key child collections by composite routing key to fix shared correlation key collision

When multiple parents share the same correlation key but have different
parent-referenced filter values, child collections were incorrectly shared.
Fix by keying child collections by (correlationKey, parentFilterValues)
composite, and using composite child keys in the D2 stream to prevent
collisions.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: kevin-dp

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

@@ -38,6 +38,9 @@ import type { QueryCache, QueryMapping, WindowOptions } from './types.js'
 
 export type { WindowOptions } from './types.js'
 
+/** Symbol used to tag parent $selected with routing metadata for includes */
+export const INCLUDES_ROUTING = Symbol(`includesRouting`)
+
 /**
  * Result of compiling an includes subquery, including the child pipeline
  * and metadata needed to route child results to parent-scoped Collections.
@@ -55,6 +58,8 @@ export interface IncludesCompilationResult {
   hasOrderBy: boolean
   /** Full compilation result for the child query (for nested includes + alias tracking) */
   childCompilationResult: CompilationResult
+  /** Parent-side projection refs for parent-referencing filters */
+  parentProjection?: Array<PropRef>
 }
 
 /**
@@ -213,7 +218,11 @@ export function compileQuery(
         if (parentSide != null) {
           tagged.__parentContext = parentSide
         }
-        return [childKey, tagged]
+        const effectiveKey =
+          parentSide != null
+            ? `${String(childKey)}::${JSON.stringify(parentSide)}`
+            : childKey
+        return [effectiveKey, tagged]
       }),
     )
 
@@ -231,6 +240,7 @@ export function compileQuery(
       const nsRow: Record<string, any> = { [mainSource]: cleanRow }
       if (__parentContext) {
         Object.assign(nsRow, __parentContext)
+        ;(nsRow as any).__parentContext = __parentContext
       }
       const ret = [key, nsRow] as [string, Record<string, typeof row>]
       return ret
@@ -291,6 +301,13 @@ export function compileQuery(
   // This must happen AFTER WHERE (so parent pipeline is filtered) but BEFORE processSelect
   // (so IncludesSubquery nodes are stripped before select compilation).
   const includesResults: Array<IncludesCompilationResult> = []
+  const includesRoutingFns: Array<{
+    fieldName: string
+    getRouting: (nsRow: any) => {
+      correlationKey: unknown
+      parentContext: Record<string, any> | null
+    }
+  }> = []
   if (query.select) {
     const includesEntries = extractIncludesFromSelect(query.select)
     for (const { key, subquery } of includesEntries) {
@@ -374,30 +391,52 @@ export function compileQuery(
           subquery.query.orderBy && subquery.query.orderBy.length > 0
         ),
         childCompilationResult: childResult,
+        parentProjection: subquery.parentProjection,
       })
 
+      // Capture routing function for INCLUDES_ROUTING tagging
+      if (subquery.parentProjection && subquery.parentProjection.length > 0) {
+        const compiledProjs = subquery.parentProjection.map((ref) => ({
+          alias: ref.path[0]!,
+          field: ref.path.slice(1),
+          compiled: compileExpression(ref),
+        }))
+        const compiledCorr = compiledCorrelation
+        includesRoutingFns.push({
+          fieldName: subquery.fieldName,
+          getRouting: (nsRow: any) => {
+            const parentContext: Record<string, Record<string, any>> = {}
+            for (const proj of compiledProjs) {
+              if (!parentContext[proj.alias]) {
+                parentContext[proj.alias] = {}
+              }
+              const value = proj.compiled(nsRow)
+              let target = parentContext[proj.alias]!
+              for (let i = 0; i < proj.field.length - 1; i++) {
+                if (!target[proj.field[i]!]) {
+                  target[proj.field[i]!] = {}
+                }
+                target = target[proj.field[i]!]
+              }
+              target[proj.field[proj.field.length - 1]!] = value
+            }
+            return { correlationKey: compiledCorr(nsRow), parentContext }
+          },
+        })
+      } else {
+        includesRoutingFns.push({
+          fieldName: subquery.fieldName,
+          getRouting: (nsRow: any) => ({
+            correlationKey: compiledCorrelation(nsRow),
+            parentContext: null,
+          }),
+        })
+      }
+
       // Replace includes entry in select with a null placeholder
       replaceIncludesInSelect(query.select, key)
     }
 
-    // Stamp correlation key values onto the namespaced row so they survive
-    // select extraction. This allows flushIncludesState to read them directly
-    // without requiring the correlation field to be in the user's select.
-    if (includesEntries.length > 0) {
-      const compiledCorrelations = includesEntries.map(({ subquery }) => ({
-        fieldName: subquery.fieldName,
-        compiled: compileExpression(subquery.correlationField),
-      }))
-      pipeline = pipeline.pipe(
-        map(([key, nsRow]: any) => {
-          const correlationKeys: Record<string, unknown> = {}
-          for (const { fieldName: fn, compiled } of compiledCorrelations) {
-            correlationKeys[fn] = compiled(nsRow)
-          }
-          return [key, { ...nsRow, __includesCorrelationKeys: correlationKeys }]
-        }),
-      )
-    }
   }
 
   if (query.distinct && !query.fnSelect && !query.select) {
@@ -442,6 +481,25 @@ export function compileQuery(
     )
   }
 
+  // Tag $selected with routing metadata for includes.
+  // This lets collection-config-builder extract routing info (correlationKey + parentContext)
+  // from parent results without depending on the user's select.
+  if (includesRoutingFns.length > 0) {
+    pipeline = pipeline.pipe(
+      map(([key, namespacedRow]: any) => {
+        const routing: Record<
+          string,
+          { correlationKey: unknown; parentContext: Record<string, any> | null }
+        > = {}
+        for (const { fieldName, getRouting } of includesRoutingFns) {
+          routing[fieldName] = getRouting(namespacedRow)
+        }
+        namespacedRow.$selected[INCLUDES_ROUTING] = routing
+        return [key, namespacedRow]
+      }),
+    )
+  }
+
   // Process the GROUP BY clause if it exists
   if (query.groupBy && query.groupBy.length > 0) {
     pipeline = processGroupBy(
@@ -531,16 +589,14 @@ export function compileQuery(
         // Extract the final results from $selected and include orderBy index
         const raw = (row as any).$selected
         const finalResults = unwrapValue(raw)
-        // Stamp includes correlation keys onto the result for child routing
-        if ((row as any).__includesCorrelationKeys) {
-          finalResults.__includesCorrelationKeys = (
-            row as any
-          ).__includesCorrelationKeys
-        }
-        // When in includes mode, embed the correlation key as third element
+        // When in includes mode, embed the correlation key and parentContext
         if (parentKeyStream) {
           const correlationKey = (row as any)[mainSource]?.__correlationKey
-          return [key, [finalResults, orderByIndex, correlationKey]] as any
+          const parentContext = (row as any).__parentContext ?? null
+          return [
+            key,
+            [finalResults, orderByIndex, correlationKey, parentContext],
+          ] as any
         }
         return [key, [finalResults, orderByIndex]] as [unknown, [any, string]]
       }),
@@ -570,16 +626,14 @@ export function compileQuery(
       // Extract the final results from $selected and return [key, [results, undefined]]
       const raw = (row as any).$selected
       const finalResults = unwrapValue(raw)
-      // Stamp includes correlation keys onto the result for child routing
-      if ((row as any).__includesCorrelationKeys) {
-        finalResults.__includesCorrelationKeys = (
-          row as any
-        ).__includesCorrelationKeys
-      }
-      // When in includes mode, embed the correlation key as third element
+      // When in includes mode, embed the correlation key and parentContext
       if (parentKeyStream) {
         const correlationKey = (row as any)[mainSource]?.__correlationKey
-        return [key, [finalResults, undefined, correlationKey]] as any
+        const parentContext = (row as any).__parentContext ?? null
+        return [
+          key,
+          [finalResults, undefined, correlationKey, parentContext],
+        ] as any
       }
       return [key, [finalResults, undefined]] as [
         unknown,

packages/db/src/query/live/collection-config-builder.ts

@@ -1,5 +1,5 @@
 import { D2, output, serializeValue } from '@tanstack/db-ivm'
-import { compileQuery } from '../compiler/index.js'
+import { INCLUDES_ROUTING, compileQuery } from '../compiler/index.js'
 import { createCollection } from '../../collection/index.js'
 import { IncludesSubquery } from '../ir.js'
 import { buildQuery, getQueryIR } from '../builder/index.js'
@@ -831,14 +831,21 @@ export class CollectionConfigBuilder<
           syncState.messagesCount += messages.length
 
           for (const [[childKey, tupleData], multiplicity] of messages) {
-            const [childResult, _orderByIndex, correlationKey] =
-              tupleData as unknown as [any, string | undefined, unknown]
-
-            // Accumulate by [correlationKey, childKey]
-            let byChild = state.pendingChildChanges.get(correlationKey)
+            const [childResult, _orderByIndex, correlationKey, parentContext] =
+              tupleData as unknown as [
+                any,
+                string | undefined,
+                unknown,
+                Record<string, any> | null,
+              ]
+
+            const routingKey = computeRoutingKey(correlationKey, parentContext)
+
+            // Accumulate by [routingKey, childKey]
+            let byChild = state.pendingChildChanges.get(routingKey)
             if (!byChild) {
               byChild = new Map()
-              state.pendingChildChanges.set(correlationKey, byChild)
+              state.pendingChildChanges.set(routingKey, byChild)
             }
 
             const existing = byChild.get(childKey) || {
@@ -1351,13 +1358,20 @@ function setupNestedPipelines(
         syncState.messagesCount += messages.length
 
         for (const [[childKey, tupleData], multiplicity] of messages) {
-          const [childResult, _orderByIndex, correlationKey] =
-            tupleData as unknown as [any, string | undefined, unknown]
+          const [childResult, _orderByIndex, correlationKey, parentContext] =
+            tupleData as unknown as [
+              any,
+              string | undefined,
+              unknown,
+              Record<string, any> | null,
+            ]
+
+          const routingKey = computeRoutingKey(correlationKey, parentContext)
 
-          let byChild = buffer.get(correlationKey)
+          let byChild = buffer.get(routingKey)
           if (!byChild) {
             byChild = new Map()
-            buffer.set(correlationKey, byChild)
+            buffer.set(routingKey, byChild)
           }
 
           const existing = byChild.get(childKey) || {
@@ -1499,11 +1513,12 @@ function updateRoutingIndex(
   for (const setup of state.nestedSetups) {
     for (const [, change] of childChanges) {
       if (change.inserts > 0) {
-        // Read the pre-computed nested correlation key from the compiler stamp
-        const nestedCorrelationKey =
-          change.value.__includesCorrelationKeys?.[
+        // Read the nested correlation key from the INCLUDES_ROUTING stamp
+        const nestedRouting =
+          change.value[INCLUDES_ROUTING]?.[
             setup.compilationResult.fieldName
           ]
+        const nestedCorrelationKey = nestedRouting?.correlationKey
 
         if (nestedCorrelationKey != null) {
           state.nestedRoutingIndex!.set(nestedCorrelationKey, correlationKey)
@@ -1516,10 +1531,11 @@ function updateRoutingIndex(
         }
       } else if (change.deletes > 0 && change.inserts === 0) {
         // Remove from routing index
-        const nestedCorrelationKey =
-          change.value.__includesCorrelationKeys?.[
+        const nestedRouting2 =
+          change.value[INCLUDES_ROUTING]?.[
             setup.compilationResult.fieldName
           ]
+        const nestedCorrelationKey = nestedRouting2?.correlationKey
 
         if (nestedCorrelationKey != null) {
           state.nestedRoutingIndex!.delete(nestedCorrelationKey)
@@ -1568,6 +1584,19 @@ function hasNestedBufferChanges(setups: Array<NestedIncludesSetup>): boolean {
   return false
 }
 
+/**
+ * Computes a composite routing key from correlation key and parent context.
+ * When parentContext is null (no parent filters), returns the raw correlationKey
+ * for zero behavioral change on existing queries.
+ */
+function computeRoutingKey(
+  correlationKey: unknown,
+  parentContext: Record<string, any> | null,
+): unknown {
+  if (parentContext == null) return correlationKey
+  return JSON.stringify([correlationKey, parentContext])
+}
+
 /**
  * Creates a child Collection entry for includes subqueries.
  * The child Collection is a full-fledged Collection instance that starts syncing immediately.
@@ -1640,33 +1669,35 @@ function flushIncludesState(
       for (const [parentKey, changes] of parentChanges) {
         if (changes.inserts > 0) {
           const parentResult = changes.value
-          // Read the pre-computed correlation key from the compiler stamp
-          const correlationKey =
-            parentResult.__includesCorrelationKeys?.[state.fieldName]
+          // Extract routing info from INCLUDES_ROUTING symbol (set by compiler)
+          const routing = parentResult[INCLUDES_ROUTING]?.[state.fieldName]
+          const correlationKey = routing?.correlationKey
+          const parentContext = routing?.parentContext ?? null
+          const routingKey = computeRoutingKey(correlationKey, parentContext)
 
           if (correlationKey != null) {
-            // Ensure child Collection exists for this correlation key
-            if (!state.childRegistry.has(correlationKey)) {
+            // Ensure child Collection exists for this routing key
+            if (!state.childRegistry.has(routingKey)) {
               const entry = createChildCollectionEntry(
                 parentId,
                 state.fieldName,
-                correlationKey,
+                routingKey,
                 state.hasOrderBy,
                 state.nestedSetups,
               )
-              state.childRegistry.set(correlationKey, entry)
+              state.childRegistry.set(routingKey, entry)
             }
-            // Update reverse index: correlation key → parent keys
-            let parentKeys = state.correlationToParentKeys.get(correlationKey)
+            // Update reverse index: routing key → parent keys
+            let parentKeys = state.correlationToParentKeys.get(routingKey)
             if (!parentKeys) {
               parentKeys = new Set()
-              state.correlationToParentKeys.set(correlationKey, parentKeys)
+              state.correlationToParentKeys.set(routingKey, parentKeys)
             }
             parentKeys.add(parentKey)
 
             // Attach child Collection to the parent result
             parentResult[state.fieldName] =
-              state.childRegistry.get(correlationKey)!.collection
+              state.childRegistry.get(routingKey)!.collection
           }
         }
       }
@@ -1769,17 +1800,21 @@ function flushIncludesState(
     if (parentChanges) {
       for (const [parentKey, changes] of parentChanges) {
         if (changes.deletes > 0 && changes.inserts === 0) {
-          const correlationKey =
-            changes.value.__includesCorrelationKeys?.[state.fieldName]
+          const routing =
+            changes.value[INCLUDES_ROUTING]?.[state.fieldName]
+          const correlationKey = routing?.correlationKey
+          const parentContext = routing?.parentContext ?? null
+          const routingKey = computeRoutingKey(correlationKey, parentContext)
           if (correlationKey != null) {
-            cleanRoutingIndexOnDelete(state, correlationKey)
-            state.childRegistry.delete(correlationKey)
+            cleanRoutingIndexOnDelete(state, routingKey)
+            state.childRegistry.delete(routingKey)
             // Clean up reverse index
-            const parentKeys = state.correlationToParentKeys.get(correlationKey)
+            const parentKeys =
+              state.correlationToParentKeys.get(routingKey)
             if (parentKeys) {
               parentKeys.delete(parentKey)
               if (parentKeys.size === 0) {
-                state.correlationToParentKeys.delete(correlationKey)
+                state.correlationToParentKeys.delete(routingKey)
               }
             }
           }
@@ -1788,10 +1823,10 @@ function flushIncludesState(
     }
   }
 
-  // Clean up the internal stamp from parent/child results so it doesn't leak to the user
+  // Clean up the internal routing stamp from parent/child results
   if (parentChanges) {
     for (const [, changes] of parentChanges) {
-      delete changes.value.__includesCorrelationKeys
+      delete changes.value[INCLUDES_ROUTING]
     }
   }
 }