feat(databases): Add support for fast bulk operations

daffl · daffl · commit fcc49aaf13ee · 2022-12-19T21:24:21.000-08:00
diff --git a/docs/api/databases/common.md b/docs/api/databases/common.md
@@ -102,6 +102,18 @@ Disabling or changing the default pagination is not available in the client. Onl
 
 </BlockQuote>
 
+## Bulk updates
+
+Some database adapters allow to set the `params.bulk` option to perform fast `create`, `patch` or `remove` operations for a large amount of data. Setting `params.bulk = true` will always return no data (an empty array `[]`) and not send any real-time events.
+
+```ts
+const manyTodos = await readCSVFile('todos.csv')
+
+await app.service('todos').create(manyTodos, {
+  bulk: true
+}) // -> []
+```
+
 ## Extending Adapters
 
 There are two ways to extend existing database adapters. Either by extending the base class or by adding functionality through hooks.
diff --git a/packages/adapter-commons/src/declarations.ts b/packages/adapter-commons/src/declarations.ts
@@ -70,6 +70,7 @@ export interface AdapterParams<
 > extends Params<Q> {
   adapter?: A
   paginate?: PaginationParams
+  bulk?: boolean
 }
 
 /**
diff --git a/packages/adapter-commons/src/service.ts b/packages/adapter-commons/src/service.ts
@@ -57,6 +57,10 @@ export abstract class AdapterBase<
    * @returns Wether or not multiple updates are allowed.
    */
   allowsMulti(method: string, params: ServiceParams = {} as ServiceParams) {
+    if (params.bulk) {
+      return true
+    }
+
     const always = alwaysMulti[method]
 
     if (typeof always !== 'undefined') {
diff --git a/packages/adapter-tests/src/declarations.ts b/packages/adapter-tests/src/declarations.ts
@@ -33,6 +33,7 @@ export type AdapterMethodsTestName =
   | '.remove'
   | '.remove + $select'
   | '.remove + id + query'
+  | '.remove bulk'
   | '.remove + multi'
   | '.remove + multi no pagination'
   | '.remove + id + query id'
@@ -49,12 +50,14 @@ export type AdapterMethodsTestName =
   | '.patch multiple no pagination'
   | '.patch multi query same'
   | '.patch multi query changed'
+  | '.patch bulk'
   | '.patch + NotFound'
   | '.patch + query + NotFound'
   | '.patch + id + query id'
   | '.create'
   | '.create + $select'
   | '.create multi'
+  | '.create bulk'
   | 'internal .find'
   | 'internal .get'
   | 'internal .create'
diff --git a/packages/adapter-tests/src/methods.ts b/packages/adapter-tests/src/methods.ts
@@ -114,6 +114,28 @@ export default (test: AdapterMethodsTest, app: any, _errors: any, serviceName: s
         }
       })
 
+      test('.remove bulk', async () => {
+        await service.create({ name: 'Dave', age: 29, created: true })
+        await service.create({
+          name: 'David',
+          age: 3,
+          created: true
+        })
+
+        const data = await service.remove(null, {
+          query: { created: true },
+          bulk: true
+        })
+
+        assert.deepStrictEqual(data, [])
+
+        const found = await service.find({
+          query: { created: true }
+        })
+
+        assert.strictEqual(found.length, 0)
+      })
+
       test('.remove + multi', async () => {
         try {
           await service.remove(null)
@@ -398,6 +420,39 @@ export default (test: AdapterMethodsTest, app: any, _errors: any, serviceName: s
         await service.remove(david[idProp])
       })
 
+      test('.patch bulk', async () => {
+        const dave = await service.create({
+          name: 'Dave',
+          age: 29,
+          created: true
+        })
+        const david = await service.create({
+          name: 'David',
+          age: 3,
+          created: true
+        })
+
+        const data = await service.patch(
+          null,
+          {
+            age: 2
+          },
+          {
+            query: { created: true },
+            bulk: true
+          }
+        )
+
+        assert.deepStrictEqual(data, [])
+
+        const daveAfter = await service.get(dave[idProp])
+
+        assert.strictEqual(daveAfter.age, 2, 'Dave age was updated')
+
+        await service.remove(dave[idProp])
+        await service.remove(david[idProp])
+      })
+
       test('.patch multiple no pagination', async () => {
         try {
           await service.remove(doug[idProp])
@@ -643,6 +698,36 @@ export default (test: AdapterMethodsTest, app: any, _errors: any, serviceName: s
         await service.remove(data[0][idProp])
         await service.remove(data[1][idProp])
       })
+
+      test('.create bulk', async () => {
+        const items = [
+          {
+            name: 'Gerald',
+            age: 18
+          },
+          {
+            name: 'Herald',
+            age: 18
+          }
+        ]
+
+        const data = await service.create(items, {
+          bulk: true
+        })
+
+        assert.deepStrictEqual(data, [])
+
+        const foundItems = await service.find({
+          query: { age: 18 }
+        })
+
+        assert.strictEqual(foundItems.length, 2)
+
+        await service.remove(null, {
+          query: { age: 18 },
+          bulk: true
+        })
+      })
     })
 
     describe("doesn't call public methods internally", () => {
diff --git a/packages/knex/src/adapter.ts b/packages/knex/src/adapter.ts
@@ -223,6 +223,15 @@ export class KnexAdapter<
   ): Promise<Result | Result[]> {
     const data = _data as any
 
+    if (params.bulk) {
+      const res = await this.db(params)
+        .insert(data)
+        .then(() => [])
+        .catch(errorHandler)
+
+      return res as Result[]
+    }
+
     if (Array.isArray(data)) {
       return Promise.all(data.map((current) => this._create(current, params)))
     }
@@ -252,6 +261,12 @@ export class KnexAdapter<
     }
 
     const data = _.omit(raw, this.id)
+
+    if (params.bulk) {
+      await this.createQuery(params).update(data)
+      return []
+    }
+
     const results = await this._findOrGet(id, {
       ...params,
       query: {
@@ -313,6 +328,11 @@ export class KnexAdapter<
       throw new MethodNotAllowed('Can not remove multiple entries')
     }
 
+    if (params.bulk) {
+      await this.createQuery(params).del().catch(errorHandler)
+      return []
+    }
+
     const items = await this._findOrGet(id, params)
     const { query } = this.filterQuery(params)
     const q = this.db(params)
diff --git a/packages/knex/test/index.test.ts b/packages/knex/test/index.test.ts
@@ -27,6 +27,7 @@ const testSuite = adapterTests([
   '.remove + id + query',
   '.remove + multi',
   '.remove + multi no pagination',
+  '.remove bulk',
   '.remove + id + query id',
   '.update',
   '.update + $select',
@@ -38,6 +39,7 @@ const testSuite = adapterTests([
   '.patch + $select',
   '.patch + id + query',
   '.patch multiple',
+  '.patch bulk',
   '.patch multiple no pagination',
   '.patch multi query same',
   '.patch multi query changed',
@@ -47,6 +49,7 @@ const testSuite = adapterTests([
   '.create',
   '.create + $select',
   '.create multi',
+  '.create bulk',
   'internal .find',
   'internal .get',
   'internal .create',
diff --git a/packages/memory/src/index.ts b/packages/memory/src/index.ts
@@ -6,7 +6,8 @@ import {
   AdapterBase,
   AdapterServiceOptions,
   PaginationOptions,
-  AdapterParams
+  AdapterParams,
+  AdapterQuery
 } from '@feathersjs/adapter-commons'
 import sift from 'sift'
 import { NullableId, Id, Params, Paginated } from '@feathersjs/feathers'
@@ -28,10 +29,12 @@ const _select = (data: any, params: any, ...args: any[]) => {
   return base(JSON.parse(JSON.stringify(data)))
 }
 
+export type MemoryAdapterParams<Q = AdapterQuery> = AdapterParams<Q, Partial<MemoryServiceOptions>>
+
 export class MemoryAdapter<
   Result = any,
   Data = Partial<Result>,
-  ServiceParams extends Params = Params,
+  ServiceParams extends MemoryAdapterParams = MemoryAdapterParams,
   PatchData = Partial<Data>
 > extends AdapterBase<Result, Data, PatchData, ServiceParams, MemoryServiceOptions<Result>> {
   store: MemoryServiceStore<Result>
@@ -145,18 +148,18 @@ export class MemoryAdapter<
   async _create(data: Partial<Data>[], params?: ServiceParams): Promise<Result[]>
   async _create(data: Partial<Data> | Partial<Data>[], _params?: ServiceParams): Promise<Result | Result[]>
   async _create(
-    data: Partial<Data> | Partial<Data>[],
+    _data: Partial<Data> | Partial<Data>[],
     params: ServiceParams = {} as ServiceParams
   ): Promise<Result | Result[]> {
-    if (Array.isArray(data)) {
-      return Promise.all(data.map((current) => this._create(current, params)))
-    }
+    const payload = Array.isArray(_data) ? _data : [_data]
+    const results = payload.map((value) => {
+      const id = (value as any)[this.id] || this._uId++
+      const current = _.extend({}, value, { [this.id]: id })
 
-    const id = (data as any)[this.id] || this._uId++
-    const current = _.extend({}, data, { [this.id]: id })
-    const result = (this.store[id] = current)
+      return _select((this.store[id] = current), params, this.id)
+    })
 
-    return _select(result, params, this.id) as Result
+    return params.bulk ? [] : Array.isArray(_data) ? results : results[0]
   }
 
   async _update(id: Id, data: Data, params: ServiceParams = {} as ServiceParams): Promise<Result> {
@@ -202,11 +205,12 @@ export class MemoryAdapter<
         ...params,
         query
       })
+      const results = entries.map(patchEntry)
 
-      return entries.map(patchEntry)
+      return params.bulk ? [] : results
     }
 
-    return patchEntry(await this._get(id, params)) // Will throw an error if not found
+    return params.bulk ? [] : patchEntry(await this._get(id, params)) // Will throw an error if not found
   }
 
   async _remove(id: null, params?: ServiceParams): Promise<Result[]>
@@ -225,7 +229,9 @@ export class MemoryAdapter<
         query
       })
 
-      return Promise.all(entries.map((current: any) => this._remove(current[this.id] as Id, params)))
+      entries.forEach((current: any) => delete this.store[(current as any)[this.id]])
+
+      return params.bulk ? [] : entries
     }
 
     const entry = await this._get(id, params)
diff --git a/packages/memory/test/index.test.ts b/packages/memory/test/index.test.ts
@@ -26,6 +26,7 @@ const testSuite = adapterTests([
   '.remove + id + query',
   '.remove + multi',
   '.remove + multi no pagination',
+  '.remove bulk',
   '.remove + id + query id',
   '.update',
   '.update + $select',
@@ -40,12 +41,14 @@ const testSuite = adapterTests([
   '.patch multiple no pagination',
   '.patch multi query same',
   '.patch multi query changed',
+  '.patch bulk',
   '.patch + query + NotFound',
   '.patch + NotFound',
   '.patch + id + query id',
   '.create',
   '.create + $select',
   '.create multi',
+  '.create bulk',
   'internal .find',
   'internal .get',
   'internal .create',
diff --git a/packages/mongodb/src/adapter.ts b/packages/mongodb/src/adapter.ts
@@ -261,7 +261,6 @@ export class MongoDbAdapter<
     data: Data | Data[],
     params: ServiceParams = {} as ServiceParams
   ): Promise<Result | Result[]> {
-    const writeOptions = params.mongodb
     const model = await this.getModel(params)
     const setId = (item: any) => {
       const entry = Object.assign({}, item)
@@ -279,14 +278,14 @@ export class MongoDbAdapter<
 
     const promise = Array.isArray(data)
       ? model
-          .insertMany(data.map(setId), writeOptions)
+          .insertMany(data.map(setId), params.mongodb)
           .then(async (result) =>
-            Promise.all(
-              Object.values(result.insertedIds).map(async (_id) => model.findOne({ _id }, params.mongodb))
-            )
+            params.bulk
+              ? []
+              : model.find({ _id: { $in: Object.values(result.insertedIds) } }, params.mongodb).toArray()
           )
       : model
-          .insertOne(setId(data), writeOptions)
+          .insertOne(setId(data), params.mongodb)
           .then(async (result) => model.findOne({ _id: result.insertedId }, params.mongodb))
 
     return promise.then(select(params, this.id)).catch(errorHandler)
@@ -325,6 +324,12 @@ export class MongoDbAdapter<
 
       return current
     }, {} as any)
+
+    if (params.bulk) {
+      await model.updateMany(query, modifier, updateOptions)
+      return []
+    }
+
     const originalIds = await this._findOrGet(id, {
       ...params,
       query: {
@@ -389,11 +394,13 @@ export class MongoDbAdapter<
       }
     }
 
-    return this._findOrGet(id, findParams)
-      .then(async (items) => {
-        await model.deleteMany(query, deleteOptions)
-        return items
-      })
-      .catch(errorHandler)
+    return params.bulk
+      ? model.deleteMany(query, deleteOptions).then(() => [])
+      : this._findOrGet(id, findParams)
+          .then(async (items) => {
+            await model.deleteMany(query, deleteOptions)
+            return items
+          })
+          .catch(errorHandler)
   }
 }
diff --git a/packages/mongodb/test/index.test.ts b/packages/mongodb/test/index.test.ts

Original file line number	Diff line number	Diff line change
`@@ -70,6 +70,7 @@ export interface AdapterParams<`
`70`	`70`	`> extends Params<Q> {`
`71`	`71`	`adapter?: A`
`72`	`72`	`paginate?: PaginationParams`
	`73`	`+ bulk?: boolean`
`73`	`74`	`}`
`74`	`75`
`75`	`76`	`/**`