Visitar URL original
feat(utils): add `keyset` option to `iterateFind` and `chunkFind` (#78) · feathersjs/feathers-utils@ba03936 · GitHub
Skip to content

Commit ba03936

Browse files
fratzingerclaude
fratzinger
and
claude
authored
feat(utils): add keyset option to iterateFind and chunkFind (#78)
With `$skip`, the pages shift when the loop changes items out of the query or others write to the service: items are skipped or seen twice, without an error. `keyset: true` asks every page for the items after the last id of the page before, sorted by id, with `paginate: false` - every item comes once, and large tables stay linear. Both utils now share one pager, `findPages`, and document their options. Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 9c05309 commit ba03936

8 files changed

Lines changed: 425 additions & 70 deletions

File tree

‎src/common/find-pages.ts‎

Lines changed: 126 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,126 @@
1+
import type { Params } from '@feathersjs/feathers'
2+
import { addToQuery } from '../utils/add-to-query/add-to-query.util.js'
3+
import { isEqual } from './is-equal.js'
4+
5+
type PaginateOption = { default?: number; max?: number }
6+
7+
/** What `iterateFind` and `chunkFind` take - documented there. */
8+
export type FindPagesOptions<P extends Params = Params> = {
9+
params?: P & { paginate?: PaginateOption }
10+
keyset?: boolean
11+
}
12+
13+
/**
14+
* The pages of a `find`, one after the other - shared by `iterateFind` and
15+
* `chunkFind`. Pages by `$skip`, or by the id with the `keyset` option.
16+
*/
17+
export async function* findPages(
18+
service: any,
19+
options: FindPagesOptions | undefined,
20+
defaultLimit?: number,
21+
): AsyncGenerator<any[], void, unknown> {
22+
if (options?.keyset) {
23+
yield* keysetPages(service, options)
24+
return
25+
}
26+
27+
const params = {
28+
...options?.params,
29+
query: {
30+
...(options?.params?.query ?? {}),
31+
$limit: options?.params?.query?.$limit ?? defaultLimit,
32+
$skip: options?.params?.query?.$skip ?? 0,
33+
},
34+
paginate: {
35+
default: options?.params?.paginate?.default ?? 10,
36+
max: options?.params?.paginate?.max ?? 100,
37+
},
38+
}
39+
40+
let $skip = params.query.$skip
41+
let result
42+
43+
do {
44+
// a fresh params object per page: the service may hold on to the last one
45+
result = await service.find({
46+
...params,
47+
query: { ...params.query, $skip },
48+
})
49+
50+
// Guard against an infinite loop: an empty page never advances $skip, so
51+
// `total > $skip` could stay true forever (e.g. $limit:0, or a stale total
52+
// when items are concurrently removed / filtered out by hooks).
53+
if (!result.data.length) {
54+
break
55+
}
56+
57+
yield result.data
58+
59+
$skip += result.data.length
60+
} while (result.total > $skip)
61+
}
62+
63+
async function* keysetPages(
64+
service: any,
65+
options: FindPagesOptions,
66+
): AsyncGenerator<any[], void, unknown> {
67+
const idField: string = service.id ?? 'id'
68+
const { $sort, $limit, $skip, $select, ...query } =
69+
options.params?.query ?? {}
70+
71+
if ($sort && Object.keys($sort).length) {
72+
throw new Error(
73+
`keyset pagination sorts by '${idField}', so it can't take a $sort`,
74+
)
75+
}
76+
77+
const { paginate, ...params } = options.params ?? {}
78+
const limit = Math.min(
79+
$limit ?? paginate?.default ?? 10,
80+
paginate?.max ?? 100,
81+
)
82+
const select =
83+
Array.isArray($select) && !$select.includes(idField)
84+
? [...$select, idField]
85+
: $select
86+
87+
let skip: number | undefined = $skip
88+
let after: { id: unknown } | undefined
89+
90+
while (true) {
91+
let pageQuery: Record<string, any> = {
92+
...query,
93+
$sort: { [idField]: 1 },
94+
$limit: limit,
95+
...(skip !== undefined && { $skip: skip }),
96+
...(select !== undefined && { $select: select }),
97+
}
98+
if (after) {
99+
// through `addToQuery`, so a condition on the id in `query` is kept
100+
pageQuery = addToQuery(pageQuery, { [idField]: { $gt: after.id } })
101+
}
102+
103+
const page: any[] = await service.find({
104+
...params,
105+
query: pageQuery,
106+
paginate: false,
107+
})
108+
109+
if (!page.length) {
110+
return
111+
}
112+
113+
// read before the page is handed out: the loop may change its items
114+
const lastId = page[page.length - 1]?.[idField]
115+
if (lastId == null || (after && isEqual(lastId, after.id))) {
116+
throw new Error(
117+
`keyset pagination made no progress: the items need their '${idField}', sorted by \`$sort\``,
118+
)
119+
}
120+
121+
yield page
122+
123+
after = { id: lastId }
124+
skip = undefined
125+
}
126+
}

‎src/common/index.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,8 @@ export function isPromise(p: any): p is Promise<any> {
1515

1616
export { allowsMulti } from './allows-multi.js'
1717
export { changeMany } from './change-many.js'
18+
export { findPages } from './find-pages.js'
19+
export type { FindPagesOptions } from './find-pages.js'
1820
export { mapLimit, DEFAULT_CONCURRENCY } from './map-limit.js'
1921
export { traverse } from './traverse.js'
2022
export { clone } from './clone.js'

‎src/utils/chunk-find/chunk-find.util.md‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,28 @@ category: utils
44
tags:
55
- query
66
- batching
7+
options: ChunkFindOptions
78
see:
89
- utils/iterateFind
910
---
11+
12+
## Changing Items While Iterating
13+
14+
By default, `chunkFind` pages with `$skip`. When the loop changes items so that they no longer match the query, or others write to the service meanwhile, the pages shift: items are skipped or seen twice, without an error.
15+
16+
```ts
17+
for await (const jobs of chunkFind(app, 'jobs', {
18+
params: { query: { status: 'pending' } },
19+
// pages by id instead of `$skip`, so every job comes exactly once
20+
keyset: true,
21+
})) {
22+
const ids = jobs.map((job) => job.id)
23+
await app
24+
.service('jobs')
25+
.patch(null, { status: 'done' }, { query: { id: { $in: ids } } })
26+
}
27+
```
28+
29+
Without `keyset`, this loop leaves jobs `pending`: after the first page, those jobs no longer match, so `$skip` jumps over as many unprocessed ones.
30+
31+
With `keyset: true`, every page asks for the items after the last id of the page before. That also keeps large tables cheap, since the database no longer reads and discards every skipped row per page. The items come sorted by id, so `query.$sort` is not supported.

‎src/utils/chunk-find/chunk-find.util.test.ts‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -169,4 +169,51 @@ describe('chunkFind', function () {
169169
)
170170
expect(skips).toEqual([0, 2])
171171
})
172+
173+
describe('keyset', () => {
174+
it('yields the pages of `$limit`, sorted by id', async function () {
175+
const app = feathers().use(
176+
'items',
177+
new MemoryService({ startId: 1, multi: true }),
178+
)
179+
await app.service('items').create(Array.from({ length: 25 }, () => ({})))
180+
181+
const pages: number[][] = []
182+
for await (const page of chunkFind(app, 'items', {
183+
params: { query: { $limit: 10 } },
184+
keyset: true,
185+
})) {
186+
pages.push(page.map((item: any) => item.id))
187+
}
188+
189+
expect(pages.map((page) => page.length)).toEqual([10, 10, 5])
190+
expect(pages.flat()).toEqual(Array.from({ length: 25 }, (_, i) => i + 1))
191+
})
192+
193+
it('yields every item when the loop changes them out of the query', async function () {
194+
const app = feathers().use(
195+
'jobs',
196+
new MemoryService({ startId: 1, multi: true }),
197+
)
198+
const jobs = app.service('jobs')
199+
await jobs.create(
200+
Array.from({ length: 50 }, () => ({ status: 'pending' })),
201+
)
202+
203+
let processed = 0
204+
for await (const page of chunkFind(app, 'jobs', {
205+
params: { query: { status: 'pending' } },
206+
keyset: true,
207+
})) {
208+
processed += page.length
209+
await jobs.patch(
210+
null,
211+
{ status: 'done' },
212+
{ query: { id: { $in: page.map((job: any) => job.id) } } },
213+
)
214+
}
215+
216+
expect(processed).toBe(50)
217+
})
218+
})
172219
})

‎src/utils/chunk-find/chunk-find.util.ts‎

Lines changed: 23 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import type { Application, Params } from '@feathersjs/feathers'
2+
import { findPages } from '../../common/index.js'
23
import type { KeyOf } from '../../internal.utils.js'
34
import type {
45
InferFindParams,
@@ -7,8 +8,28 @@ import type {
78

89
type PaginateOption = { default?: number; max?: number }
910

10-
type ChunkFindOptions<P extends Params = Params> = {
11+
export type ChunkFindOptions<P extends Params = Params> = {
12+
/**
13+
* The params of every `find` call. `query.$limit` is the page size, capped by
14+
* `paginate.max`; `paginate` defaults to `{ default: 10, max: 100 }`.
15+
*/
1116
params?: P & { paginate?: PaginateOption }
17+
/**
18+
* Page by the id instead of by `$skip`: every page asks for the items after
19+
* the last id of the page before, sorted by id.
20+
*
21+
* Use it when the loop changes items so that they leave the query, or when
22+
* others write to the service meanwhile - with `$skip` the pages shift, so
23+
* items are skipped or seen twice. It also stays linear on large tables,
24+
* where the database reads and discards every skipped row per page, and it
25+
* needs no `total`, so the `find` calls run with `paginate: false`.
26+
*
27+
* The items come sorted by id, so a `query.$sort` throws. The id is added to
28+
* a `query.$select` that leaves it out.
29+
*
30+
* @default false
31+
*/
32+
keyset?: boolean
1233
}
1334

1435
/**
@@ -50,38 +71,5 @@ export async function* chunkFind<
5071
throw new Error(`Service '${servicePath}' does not have a 'find' method.`)
5172
}
5273

53-
const params = {
54-
...options?.params,
55-
query: {
56-
...(options?.params?.query ?? {}),
57-
$limit: options?.params?.query?.$limit ?? 10,
58-
$skip: options?.params?.query?.$skip ?? 0,
59-
},
60-
paginate: {
61-
default: options?.params?.paginate?.default ?? 10,
62-
max: options?.params?.paginate?.max ?? 100,
63-
},
64-
}
65-
66-
let $skip = params.query.$skip
67-
let result
68-
69-
do {
70-
// a fresh params object per page: the service may hold on to the last one
71-
result = await (service as any).find({
72-
...params,
73-
query: { ...params.query, $skip },
74-
})
75-
76-
// Guard against an infinite loop: an empty page never advances $skip, so
77-
// `total > $skip` could stay true forever (e.g. $limit:0, or a stale total
78-
// when items are concurrently removed / filtered out by hooks).
79-
if (!result.data.length) {
80-
break
81-
}
82-
83-
yield result.data
84-
85-
$skip += result.data.length
86-
} while (result.total > $skip)
74+
yield* findPages(service, options, 10)
8775
}

‎src/utils/iterate-find/iterate-find.util.md‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,25 @@ category: utils
44
tags:
55
- query
66
- batching
7+
options: IterateFindOptions
78
see:
89
- utils/chunkFind
910
---
11+
12+
## Changing Items While Iterating
13+
14+
By default, `iterateFind` pages with `$skip`. When the loop changes items so that they no longer match the query, or others write to the service meanwhile, the pages shift: items are skipped or seen twice, without an error.
15+
16+
```ts
17+
for await (const job of iterateFind(app, 'jobs', {
18+
params: { query: { status: 'pending' } },
19+
// pages by id instead of `$skip`, so every job comes exactly once
20+
keyset: true,
21+
})) {
22+
await app.service('jobs').patch(job.id, { status: 'done' })
23+
}
24+
```
25+
26+
Without `keyset`, this loop leaves jobs `pending`: after the first page, those jobs no longer match, so `$skip` jumps over as many unprocessed ones.
27+
28+
With `keyset: true`, every page asks for the items after the last id of the page before. That also keeps large tables cheap, since the database no longer reads and discards every skipped row per page. The items come sorted by id, so `query.$sort` is not supported.

0 commit comments

Comments
 (0)