packages/query-core/src/streamedQuery.ts (1)

42-49: Type-safety hole: callers can specify TData without a reducer. Add overloads to prevent mismatches.

As written, a caller can do streamedQuery<number, Record<string, boolean>>({ queryFn }) with no reducer. Types accept it, but runtime returns an Array while types claim a Record. Add overloads so TData is only specifiable when reducer is provided; otherwise TData is fixed to Array.

+// Overloads to prevent specifying a custom TData without a reducer
+export function streamedQuery<
+  TQueryFnData = unknown,
+  TQueryKey extends QueryKey = QueryKey,
+>(
+  params: SimpleStreamedQueryParams<TQueryFnData, TQueryKey>
+): QueryFunction<Array<TQueryFnData>, TQueryKey>
+
+export function streamedQuery<
+  TQueryFnData = unknown,
+  TData = Array<TQueryFnData>,
+  TQueryKey extends QueryKey = QueryKey,
+>(
+  params: ReducibleStreamedQueryParams<TQueryFnData, TData, TQueryKey>
+): QueryFunction<TData, TQueryKey>
+
 export function streamedQuery<
   TQueryFnData = unknown,
   TData = Array<TQueryFnData>,
   TQueryKey extends QueryKey = QueryKey,
 >({
   queryFn,
   refetchMode = 'reset',
   reducer = (items, chunk) => addToEnd(items as Array<TQueryFnData>, chunk) as TData,
   initialValue = [] as TData,
-}: StreamedQueryParams<TQueryFnData, TData, TQueryKey>): QueryFunction<
-  TData,
-  TQueryKey
-> {
+}: StreamedQueryParams<TQueryFnData, TData, TQueryKey>): QueryFunction<TData, TQueryKey> {

Optional: move the type aliases next to this for discoverability.

docs/reference/streamedQuery.md (1)

36-44: Docs/code mismatch: use “initialValue” instead of “placeholderData”, and document requirement when a reducer is provided.

Code uses initialValue and makes it mandatory with a reducer. Update docs accordingly.

-- `reducer?: (accumulator: TData, chunk: TQueryFnData) => TData`
-  - Optional
-  - A function to reduce the streamed chunks into the final data.
-  - Defaults to a function that appends chunks to the end of the array.
-- `placeholderData?: TData = TQueryFnData`
-  - Optional
-  - Defines the initial data to be used while the first chunk is being fetched.
-  - Defaults to an empty array.
+- `reducer?: (accumulator: TData, chunk: TQueryFnData) => TData`
+  - Optional
+  - Reduces streamed chunks into the accumulator.
+  - Defaults to appending chunks to the end of an Array<TQueryFnData>.
+- `initialValue?: TData`
+  - Optional without a reducer; required when a reducer is provided.
+  - The initial accumulator used before the first chunk arrives.
+  - Defaults to an empty array when no reducer is provided.

.vscode/query.code-workspace (1)

28-29: Nit: remove stray blank lines to keep the workspace file minimal.

No functional impact, just trims noise in diffs.

-		
 		"window.zoomLevel": 2,
 		"typescript.format.enable": false,
@@
-		
 	}
 }

Also applies to: 39-41

packages/query-core/src/streamedQuery.ts (2)

26-35: JSDoc still implies Array-only data; clarify that TData can be any accumulator.

Update wording so it doesn’t promise “Data will be an Array…”. With custom reducers, TData is arbitrary.

- * Data will be an Array of all the chunks received.
+ * Data will be accumulated into TData. By default, chunks are appended to an Array<TQueryFnData>.

---

79-81: Aborted reset-refetch before first chunk can return undefined despite non-null assertion.

If refetchMode==='reset' and the stream is aborted before yielding a chunk, getQueryData(...) is undefined but asserted with !. Either (a) widen return to TData | undefined, or (b) explicitly return query?.state.data to reflect cache state (still undefined), or (c) guard with an invariant/error.

Minimal change:

-    return context.client.getQueryData(context.queryKey)!
+    return context.client.getQueryData(context.queryKey) as TData

Or, if you can accept a type change:

-}: StreamedQueryParams<...>): QueryFunction<TData, TQueryKey> {
+}: StreamedQueryParams<...>): QueryFunction<TData | undefined, TQueryKey> {
...
-    return context.client.getQueryData(context.queryKey)!
+    return context.client.getQueryData(context.queryKey)

Please confirm expected behavior for “reset + aborted before first chunk”.

Also applies to: 91-91

packages/query-core/src/__tests__/streamedQuery.test.tsx (1)

131-181: Consider adding a test for “replace + abort” to lock in cache semantics.

Edge case: refetchMode 'replace' + abort mid-stream should retain old data and never write partials. A focused test would prevent regressions.

I can add a vitest like:

test('should keep old data on replace when aborted before completion', async () => {
  // arrange: stream 2 items, then abort before the 2nd
  // expect: data stays at old [0,1], never becomes [100] mid-flight, ends with [0,1]
})

Also applies to: 240-296

docs/reference/streamedQuery.md (2)

25-29: Typing in docs: queryFn should return AsyncIterable, not TData.

Align with implementation and types.

-- `queryFn: (context: QueryFunctionContext) => Promise<AsyncIterable<TData>>`
+- `queryFn: (context: QueryFunctionContext) => Promise<AsyncIterable<TQueryFnData>>`

---

29-35: Typo: missing closing quote in refetchMode union.

Small formatting fix.

-- `refetchMode?: 'append' | 'reset' | 'replace`
+- `refetchMode?: 'append' | 'reset' | 'replace'`

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

docs/reference/streamedQuery.md (2)

36-45: Tighten wording and fix minor grammar nits

• “the streamed chunks into the final data” → fine, but remove extra spaces.
• “It is mandatory when custom reducer is provided.” → add article “a”.

-  - A function to reduce the streamed chunks into the final data.
+  - A function that reduces streamed chunks into the final data.
@@
-  - It is mandatory when custom `reducer` is provided.
+  - It is mandatory when a custom `reducer` is provided.

---

45-45: Consider adding a minimal custom reducer example

A short example will help users migrate from maxChunks and understand TQueryFnData vs TData.

+#### Example: Aggregating into a non-array accumulator
+```ts
+import { experimental_streamedQuery as streamedQuery } from '@tanstack/react-query'
+
+type Chunk = { id: string; value: number }
+type Acc = Map<string, number>
+
+const query = queryOptions({
+  queryKey: ['stats'],
+  queryFn: streamedQuery<Chunk, Acc>({
+    queryFn: fetchChunkStream, // () => Promise<AsyncIterable<Chunk>>
+    initialValue: new Map(),
+    reducer: (acc, chunk) => acc.set(chunk.id, (acc.get(chunk.id) ?? 0) + chunk.value),
+    refetchMode: 'replace',
+  }),
+})
+```

Would you like me to add a migration note showing how to replicate the old maxChunks behavior with a reducer?

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

docs/reference/streamedQuery.md (1)

36-40: Clarification on default reducer looks good

The notes now correctly state the default only applies to array TData and that non-array TData requires a custom reducer.

docs/reference/streamedQuery.md (1)

29-35: Close the missing quote in refetchMode union

Minor typo:

-- `refetchMode?: 'append' | 'reset' | 'replace`
+- `refetchMode?: 'append' | 'reset' | 'replace'`

packages/query-core/src/__tests__/streamedQuery.test.tsx (2)

417-426: Assert mid-stream accumulator to prove incremental reduction

Add an intermediate assertion to verify reducer updates between chunks:

-    await vi.advanceTimersByTimeAsync(100)
+    await vi.advanceTimersByTimeAsync(50)
+
+    expect(observer.getCurrentResult()).toMatchObject({
+      status: 'success',
+      fetchStatus: 'fetching',
+      data: { 0: true },
+    })
+
+    await vi.advanceTimersByTimeAsync(50)

---

431-445: Rename test to emphasize seed semantics (optional)

The term “initialValue (seed)” helps readers connect this to reducer semantics.

-  test('should support custom reducer with initialValue', async () => {
+  test('should support custom reducer with initialValue (seed)', async () => {

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

packages/query-core/src/streamedQuery.ts (2)

42-45: Update JSDoc to match reducer/initialValue semantics

Reflect accumulator semantics and the array-only default.

- * @param reducer - A function to reduce the streamed chunks into the final data.
- * Defaults to a function that appends chunks to the end of the array.
- * @param initialValue - Initial value to be used while the first chunk is being fetched.
+ * @param reducer - Reduces streamed chunks (`TQueryFnData`) into the accumulator (`TData`).
+ * Defaults to appending each chunk when `TData` is an array.
+ * @param initialValue - Seed value for the reducer (not shown as placeholder data).

---

85-90: Avoid double reduction per chunk and keep result in lockstep with the cache

You currently compute the reduction twice (once in setQueryData, once for result). Merge them to reduce work and prevent divergence.

-        context.client.setQueryData<TData>(context.queryKey, (prev) =>
-          reducer(prev ?? initialValue, chunk),
-        )
-      }
-      result = reducer(result, chunk)
+        context.client.setQueryData<TData>(context.queryKey, (prev) => {
+          const next = reducer((prev ?? initialValue) as TData, chunk)
+          result = next
+          return next
+        })
+      } else {
+        result = reducer(result, chunk)
+      }

packages/query-core/src/__tests__/streamedQuery.test.tsx (1)

431-471: Seeded accumulator test LGTM; consider adding empty-stream and replace+reducer cases

• Add a test where the stream yields zero chunks to lock in the fallback result.
• Add a “refetchMode: 'replace' + custom reducer” test to assert no intermediate cache writes.

Example additions:

test('returns seeded accumulator on empty stream', async () => {
  const key = queryKey()
  const observer = new QueryObserver(queryClient, {
    queryKey: key,
    queryFn: streamedQuery<number, Record<number, boolean>>({
      queryFn: async function* () {}, // no chunks
      reducer: (acc, chunk) => ({ ...acc, [chunk]: true }),
      initialValue: { 42: true },
    }),
  })
  observer.subscribe(vi.fn())
  await vi.advanceTimersByTimeAsync(10)
  expect(observer.getCurrentResult()).toMatchObject({
    status: 'success',
    fetchStatus: 'idle',
    data: { 42: true },
  })
})

test('replace refetch with custom reducer writes only at the end', async () => {
  const key = queryKey()
  let offset = 0
  const observer = new QueryObserver(queryClient, {
    queryKey: key,
    queryFn: streamedQuery<number, Record<number, boolean>>({
      queryFn: () => createAsyncNumberGenerator(2, offset),
      refetchMode: 'replace',
      reducer: (acc, chunk) => ({ ...acc, [chunk]: true }),
      initialValue: {},
    }),
  })
  observer.subscribe(vi.fn())
  await vi.advanceTimersByTimeAsync(100)
  expect(observer.getCurrentResult().data).toEqual({ 0: true, 1: true })
  offset = 100
  void observer.refetch()
  await vi.advanceTimersByTimeAsync(90)
  // still old data until stream ends
  expect(observer.getCurrentResult().data).toEqual({ 0: true, 1: true })
  await vi.advanceTimersByTimeAsync(10)
  expect(observer.getCurrentResult().data).toEqual({ 100: true, 101: true })
})

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.

packages/query-core/src/streamedQuery.ts (2)

85-87: Good fix: undefined check prevents “empty” collisions.

Using prev === undefined ? initialValue : prev correctly preserves null as a valid value.

---

11-31: Enforce array TData when no reducer (add overloads).

Current signature lets callers pick non-array TData in the “simple” path, while the default reducer assumes arrays. Add overloads to couple API shape to return type and close the type hole.

+// Overloads to bind API shape to TData
+export function streamedQuery<
+  TQueryFnData = unknown,
+  TQueryKey extends QueryKey = QueryKey,
+>(params: SimpleStreamedQueryParams<TQueryFnData, TQueryKey>): QueryFunction<
+  Array<TQueryFnData>,
+  TQueryKey
+>
+export function streamedQuery<
+  TQueryFnData = unknown,
+  TData = unknown,
+  TQueryKey extends QueryKey = QueryKey,
+>(params: ReducibleStreamedQueryParams<TQueryFnData, TData, TQueryKey>): QueryFunction<
+  TData,
+  TQueryKey
+>

packages/query-core/src/streamedQuery.ts (3)

46-56: Prefer resolving defaults inside the function to improve narrowing and readability.

Move defaults for reducer/initialValue out of the parameter list and resolve them locally. Keeps params in sync with the union types and avoids “phantom” defaults in the reducible path.

 export function streamedQuery<
   TQueryFnData = unknown,
   TData = Array<TQueryFnData>,
   TQueryKey extends QueryKey = QueryKey,
 >({
   queryFn,
   refetchMode = 'reset',
-  reducer = (items, chunk) =>
-    addToEnd(items as Array<TQueryFnData>, chunk) as TData,
-  initialValue = [] as TData,
+  reducer,
+  initialValue,
 }: StreamedQueryParams<TQueryFnData, TData, TQueryKey>): QueryFunction<
   TData,
   TQueryKey
 > {
-  return async (context) => {
+  return async (context) => {
+    const resolvedReducer =
+      (reducer ??
+        ((items: TData, chunk: TQueryFnData) =>
+          addToEnd(items as unknown as Array<TQueryFnData>, chunk) as TData))
+    const resolvedInitial = (initialValue ??
+      ([] as unknown as TData)) as TData

And then replace uses of reducer/initialValue below with resolvedReducer/resolvedInitial.

---

89-90: Nit: avoid reducing twice when not needed.

result is only read for replace-refetch finalization. Skip updating it otherwise to save work on hot paths.

-      result = reducer(result, chunk)
+      if (isRefetch && refetchMode === 'replace') {
+        result = reducer(result, chunk)
+      }

If you apply the “resolved*” refactor above, use resolvedReducer.

---

32-42: Heads-up: memory behavior is now fully consumer-defined.

With maxChunks removed, long-running streams can grow unbounded unless the reducer implements windowing/truncation. Call out this guidance prominently in docs with examples.

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

💡 Knowledge Base configuration:

• MCP integration is disabled by default for public repositories
• Jira integration is disabled by default for public repositories
• Linear integration is disabled by default for public repositories

You can enable these sources in your CodeRabbit configuration.