Overview

This PR extends Query Collections to support predicate pushdown from live queries by enabling multiple concurrent queries with different predicates/filters. When live queries push down predicates via loadSubset, Query Collections now create separate TanStack Query instances for each unique set of options, pass those options to both the query key builder and query function, and manage the lifecycle of multiple queries with proper reference counting and garbage collection.

Problem

When live queries push down predicates (via loadSubset), Query Collections need to:

1. Pass LoadSubsetOptions (predicates, limits, ordering) to the query key builder and query function
2. Create separate TanStack Query instances for each unique set of options
3. Track which rows belong to which queries (reference counting)
4. Only remove rows when no active queries reference them
5. Handle query garbage collection when queries are no longer needed
6. Return promises that resolve when query data is first available

Without this, Query Collections couldn't properly support the on-demand sync mode introduced in #669.

Solution

This PR implements a comprehensive multi-query management system that flows predicates from live queries through to your TanStack Query implementation:

Predicate Flow

When a live query calls loadSubset with predicates, those options flow through the system:

1. Live Query → calls collection._sync.loadSubset(options)
2. Query Collection → calls createQueryFromOpts(options)
3. Query Key Builder → receives options to create unique query key
4. Query Function → receives options via context.meta.loadSubsetOptions

Example Usage

import { createCollection } from "@tanstack/react-db"
import { queryCollectionOptions } from "@tanstack/query-db-collection"

const itemsPerPage = 20

export const todoCollection = createCollection(
  queryCollectionOptions({
    syncMode: 'on-demand', // Enable predicate pushdown
    
    // Query key builder receives LoadSubsetOptions
    queryKey: ({ limit, orderBy, where }) => {
      const page = computePageNumber(limit ?? itemsPerPage)
      return ["todos", { page, where, orderBy }]
    },

    // Query function receives options via context.meta.loadSubsetOptions
    queryFn: async (ctx) => {
      const { limit, where, orderBy } = ctx.meta?.loadSubsetOptions ?? {}
      const page = computePageNumber(limit ?? itemsPerPage)
      
      const params = new URLSearchParams({
        page: page.toString(),
        ...(where && { filter: JSON.stringify(where) }),
        ...(orderBy && { sort: JSON.stringify(orderBy) }),
      })
      
      const res = await fetch(\`/api/todos?\${params}\`)
      if (!res.ok) throw new Error("Failed to fetch todos")
      return res.json()
    },

    getKey: (item) => item.id,
    schema: todoSchema,
  })
)

function computePageNumber(limit: number): number {
  return Math.max(1, Math.ceil(limit / itemsPerPage))
}

In this example:

• The queryKey function builds different cache keys based on page/filters
• The queryFn receives the same options via ctx.meta.loadSubsetOptions to fetch the right data
• Each unique combination of predicates creates a separate TanStack Query
• Rows are reference-counted across all queries

1. Dynamic Query Keys

Type: Added TQueryKeyBuilder<TQueryKey> type

type TQueryKeyBuilder<TQueryKey> = (opts: LoadSubsetOptions) => TQueryKey

interface QueryCollectionConfig {
  queryKey: TQueryKey | TQueryKeyBuilder<TQueryKey>
  // ... other properties
}

The queryKey config option now accepts either:

• A static query key (for eager mode with no predicates)
• A function that builds a query key from LoadSubsetOptions (for on-demand mode with predicate pushdown)

LoadSubsetOptions Structure:

interface LoadSubsetOptions {
  where?: any      // Filter predicates
  orderBy?: any    // Ordering directives
  limit?: number   // Result limit
  offset?: number  // Result offset
}

2. Meta Property Extension

When creating a query, the collection merges LoadSubsetOptions into the query's meta:

const extendedMeta = { ...meta, loadSubsetOptions: opts }

const observerOptions = {
  queryKey: key,
  queryFn: queryFn,
  meta: extendedMeta,  // Contains loadSubsetOptions
  // ... other options
}

Your queryFn can then access these options via context.meta.loadSubsetOptions to fetch the appropriate data.

3. Multi-Query Tracking System

Implemented comprehensive state tracking using Maps:

// hashedQueryKey → QueryKey
const hashToQueryKey = new Map<string, QueryKey>()

// hashedQueryKey → Set<RowKey> (which rows belong to which query)
const queryToRows = new Map<string, Set<string | number>>()

// RowKey → Set<hashedQueryKey> (which queries reference each row)
const rowToQueries = new Map<string | number, Set<string>>()

// hashedQueryKey → QueryObserver (active query observers)
const observers = new Map<string, QueryObserver>()

// hashedQueryKey → unsubscribe function
const unsubscribes = new Map<string, () => void>()

Reference Counting: Rows are only deleted from the collection when their reference count drops to zero (no queries reference them anymore).

4. createQueryFromOpts Function

New internal function that creates or reuses queries based on LoadSubsetOptions:

Return Type: true | Promise<void>

• Returns true synchronously if query data is already available
• Returns Promise<void> that resolves when query data loads
• Returns Promise<void> that rejects if query encounters an error

Behavior:

• Hashes the query key (built from options) to check for existing queries
• Reuses existing QueryObserver instances when the same predicates are requested
• Creates new observers for unique predicate combinations
• Passes options to both query key builder and query function (via meta.loadSubsetOptions)
• Automatically subscribes if sync has started or collection has subscribers

5. Query Garbage Collection

Listens to TanStack Query's cache events to handle query removal:

queryClient.getQueryCache().subscribe((event) => {
  if (event.type === 'removed') {
    cleanupQuery(hashedKey)
  }
})

Cleanup Process:

1. Decrements reference counts for all rows in the removed query
2. Deletes rows when reference count reaches zero
3. Cleans up observer and unsubscribe function
4. Removes query from all tracking Maps

6. Sync Mode Integration

Eager Mode (default):

• Creates single initial query with empty options ({})
• Bypasses loadSubset (returns undefined)
• Static query key (no builder function needed)

On-Demand Mode:

• No initial query created
• Calls markReady() immediately since there's nothing to wait for
• Creates queries dynamically as loadSubset(options) is called
• Requires query key builder function to handle different predicates
• Returns createQueryFromOpts directly as the loadSubset implementation

Sync Started Tracking:
Added syncStarted flag to determine when to subscribe to new queries:

• Set to true when sync begins (via preload(), startSync, or first subscriber)
• Used instead of checking config.startSync to handle all sync scenarios

Changes

Files Modified:

• packages/query-db-collection/src/query.ts - Core implementation (+354 lines)
• packages/query-db-collection/tests/query.test.ts - Comprehensive test suite (+567 lines)
• .changeset/silent-trains-tell.md - Changeset entry

Test Coverage:

• Added 4 new tests for Query Garbage Collection scenarios
• Added test for preload() in on-demand mode
• All 64 tests passing
• Code coverage: 88.66% lines

Key Features

✅ Predicate Pushdown - Pass LoadSubsetOptions from live queries to TanStack Query
✅ Multiple Concurrent Queries - Manage multiple TanStack Query instances with different predicates
✅ Reference Counting - Track which queries reference which rows
✅ Automatic Garbage Collection - Clean up queries and rows when no longer needed
✅ Promise-Based Loading - Return promises that resolve when data is available
✅ Sync Mode Support - Works with both eager and on-demand sync modes
✅ Immediate Ready State - On-demand collections transition to ready immediately

Breaking Changes

None - this is a backward-compatible extension. Existing Query Collections with static query keys continue to work as before.

Migration Guide

If you want to enable predicate pushdown:

1. Set syncMode: 'on-demand'
2. Change queryKey from a static value to a builder function
3. Access predicates in queryFn via context.meta.loadSubsetOptions

Before:

queryCollectionOptions({
  queryKey: ['todos'],
  queryFn: async () => fetch('/api/todos').then(r => r.json()),
  // ...
})

After:

queryCollectionOptions({
  syncMode: 'on-demand',
  queryKey: ({ where, limit, orderBy }) => ['todos', { where, limit, orderBy }],
  queryFn: async (ctx) => {
    const { where, limit } = ctx.meta?.loadSubsetOptions ?? {}
    const params = new URLSearchParams()
    if (where) params.set('filter', JSON.stringify(where))
    if (limit) params.set('limit', limit.toString())
    return fetch(\`/api/todos?\${params}\`).then(r => r.json())
  },
  // ...
})

Related

• Builds on Add loadSubset State Tracking and On-Demand Sync Mode #669 (loadSubset state tracking and on-demand sync mode)
• Enables live queries to efficiently push down predicates to Query Collections
• Prepares for framework hook integration in follow-up PRs

Note: This PR replaces #646 with a rebased branch to remove duplicate commits from #669 that were already merged into main.