feat: batch mutations

Author: wschurman

packages/entity-database-adapter-knex-testing-utils/src/StubPostgresDatabaseAdapter.ts

@@ -242,75 +242,82 @@ export class StubPostgresDatabaseAdapter<
     }
   }
 
-  protected async insertInternalAsync(
+  protected async insertManyInternalAsync(
     _queryInterface: any,
     tableName: string,
-    object: object,
+    objects: readonly object[],
   ): Promise<object[]> {
     const objectCollection = this.getObjectCollectionForTable(tableName);
 
     const idField = getDatabaseFieldForEntityField(
       this.entityConfiguration2,
       this.entityConfiguration2.idField,
     );
-    const objectToInsert = {
-      [idField]: this.generateRandomID(),
-      ...object,
-    };
-    objectCollection.push(objectToInsert);
-    return [objectToInsert];
+    const insertedObjects: object[] = [];
+    for (const object of objects) {
+      const objectToInsert = {
+        [idField]: this.generateRandomID(),
+        ...object,
+      };
+      objectCollection.push(objectToInsert);
+      insertedObjects.push(objectToInsert);
+    }
+    return insertedObjects;
   }
 
-  protected async updateInternalAsync(
+  protected async updateManyInternalAsync(
     _queryInterface: any,
     tableName: string,
     tableIdField: string,
-    id: any,
-    object: object,
-  ): Promise<{ updatedRowCount: number }> {
-    // SQL does not support empty updates, mirror behavior here for better test simulation
-    if (Object.keys(object).length === 0) {
-      throw new Error(`Empty update (${tableIdField} = ${id})`);
-    }
+    items: readonly { id: any; object: object }[],
+  ): Promise<readonly { updatedRowCount: number }[]> {
+    const results: { updatedRowCount: number }[] = [];
+    for (const item of items) {
+      if (Object.keys(item.object).length === 0) {
+        throw new Error(`Empty update (${tableIdField} = ${item.id})`);
+      }
 
-    const objectCollection = this.getObjectCollectionForTable(tableName);
+      const objectCollection = this.getObjectCollectionForTable(tableName);
 
-    const objectIndex = objectCollection.findIndex((obj) => {
-      return obj[tableIdField] === id;
-    });
+      const objectIndex = objectCollection.findIndex((obj) => {
+        return obj[tableIdField] === item.id;
+      });
 
-    // SQL updates to a nonexistent row succeed but affect 0 rows,
-    // mirror that behavior here for better test simulation
-    if (objectIndex < 0) {
-      return { updatedRowCount: 0 };
-    }
+      if (objectIndex < 0) {
+        results.push({ updatedRowCount: 0 });
+        continue;
+      }
 
-    objectCollection[objectIndex] = {
-      ...objectCollection[objectIndex],
-      ...object,
-    };
-    return { updatedRowCount: 1 };
+      objectCollection[objectIndex] = {
+        ...objectCollection[objectIndex],
+        ...item.object,
+      };
+      results.push({ updatedRowCount: 1 });
+    }
+    return results;
   }
 
-  protected async deleteInternalAsync(
+  protected async deleteManyInternalAsync(
     _queryInterface: any,
     tableName: string,
     tableIdField: string,
-    id: any,
+    ids: readonly any[],
   ): Promise<number> {
     const objectCollection = this.getObjectCollectionForTable(tableName);
+    let numDeleted = 0;
 
-    const objectIndex = objectCollection.findIndex((obj) => {
-      return obj[tableIdField] === id;
-    });
+    for (const id of ids) {
+      const objectIndex = objectCollection.findIndex((obj) => {
+        return obj[tableIdField] === id;
+      });
 
-    // SQL deletes to a nonexistent row succeed and affect 0 rows,
-    // mirror that behavior here for better test simulation
-    if (objectIndex < 0) {
-      return 0;
-    }
+      if (objectIndex < 0) {
+        continue;
+      }
 
-    objectCollection.splice(objectIndex, 1);
-    return 1;
+      objectCollection.splice(objectIndex, 1);
+      numDeleted++;
+    }
+    return numDeleted;
   }
 }

packages/entity-database-adapter-knex-testing-utils/src/__tests__/StubPostgresDatabaseAdapter-test.ts

@@ -577,17 +577,19 @@ describe(StubPostgresDatabaseAdapter, () => {
     });
   });
 
-  describe('insertAsync', () => {
+  describe('insertManyAsync', () => {
     it('inserts a record', async () => {
       const queryContext = instance(mock(EntityQueryContext));
       const databaseAdapter = new StubPostgresDatabaseAdapter<TestFields, 'customIdField'>(
         testEntityConfiguration,
         new Map(),
       );
-      const result = await databaseAdapter.insertAsync(queryContext, {
-        stringField: 'hello',
-      });
-      expect(result).toMatchObject({
+      const result = await databaseAdapter.insertManyAsync(queryContext, [
+        {
+          stringField: 'hello',
+        },
+      ]);
+      expect(result[0]).toMatchObject({
         stringField: 'hello',
       });
 
@@ -602,16 +604,18 @@ describe(StubPostgresDatabaseAdapter, () => {
         testEntityConfiguration,
         new Map(),
       );
-      const result = await databaseAdapter.insertAsync(queryContext, {
-        stringField: 'hello',
-      });
+      const result = await databaseAdapter.insertManyAsync(queryContext, [
+        {
+          stringField: 'hello',
+        },
+      ]);
 
-      const ts = getTimeFromUUIDv7(result.customIdField);
+      const ts = getTimeFromUUIDv7(result[0]!.customIdField);
       expect(ts).toEqual(expectedTime);
     });
   });
 
-  describe('updateAsync', () => {
+  describe('updateManyAsync', () => {
     it('updates a record', async () => {
       const queryContext = instance(mock(EntityQueryContext));
       const databaseAdapter = new StubPostgresDatabaseAdapter<TestFields, 'customIdField'>(
@@ -635,9 +639,9 @@ describe(StubPostgresDatabaseAdapter, () => {
           ]),
         ),
       );
-      await databaseAdapter.updateAsync(queryContext, 'customIdField', 'hello', {
-        stringField: 'b',
-      });
+      await databaseAdapter.updateManyAsync(queryContext, 'customIdField', [
+        { id: 'hello', object: { stringField: 'b' } },
+      ]);
     });
 
     it('throws error when empty update to match common DBMS behavior', async () => {
@@ -664,7 +668,9 @@ describe(StubPostgresDatabaseAdapter, () => {
         ),
       );
       await expect(
-        databaseAdapter.updateAsync(queryContext, 'customIdField', 'hello', {}),
+        databaseAdapter.updateManyAsync(queryContext, 'customIdField', [
+          { id: 'hello', object: {} },
+        ]),
       ).rejects.toThrow(`Empty update (custom_id = hello)`);
     });
 
@@ -694,17 +700,17 @@ describe(StubPostgresDatabaseAdapter, () => {
 
       // Try to update a record that doesn't exist
       await expect(
-        databaseAdapter.updateAsync(
-          queryContext,
-          'customIdField',
-          'nonexistent-id', // This ID doesn't exist in the data store
-          { stringField: 'updated' },
-        ),
+        databaseAdapter.updateManyAsync(queryContext, 'customIdField', [
+          {
+            id: 'nonexistent-id', // This ID doesn't exist in the data store
+            object: { stringField: 'updated' },
+          },
+        ]),
       ).rejects.toThrow('Empty results from database adapter update');
     });
   });
 
-  describe('deleteAsync', () => {
+  describe('deleteManyAsync', () => {
     it('deletes an object', async () => {
       const queryContext = instance(mock(EntityQueryContext));
       const databaseAdapter = new StubPostgresDatabaseAdapter<TestFields, 'customIdField'>(
@@ -729,7 +735,7 @@ describe(StubPostgresDatabaseAdapter, () => {
         ),
       );
 
-      await databaseAdapter.deleteAsync(queryContext, 'customIdField', 'hello');
+      await databaseAdapter.deleteManyAsync(queryContext, 'customIdField', ['hello']);
 
       expect(
         databaseAdapter.getObjectCollectionForTable(testEntityConfiguration.tableName),
@@ -762,11 +768,9 @@ describe(StubPostgresDatabaseAdapter, () => {
 
       // Delete a record that doesn't exist
       // Unlike update, delete doesn't throw an error when the record doesn't exist
-      await databaseAdapter.deleteAsync(
-        queryContext,
-        'customIdField',
+      await databaseAdapter.deleteManyAsync(queryContext, 'customIdField', [
         'nonexistent-id', // This ID doesn't exist in the data store
-      );
+      ]);
     });
   });
 
@@ -776,21 +780,21 @@ describe(StubPostgresDatabaseAdapter, () => {
       simpleTestEntityConfiguration,
       new Map(),
     );
-    const insertedObject1 = await databaseAdapter1.insertAsync(queryContext, {});
-    expect(typeof insertedObject1.id).toBe('string');
+    const insertedObjects1 = await databaseAdapter1.insertManyAsync(queryContext, [{}]);
+    expect(typeof insertedObjects1[0]!.id).toBe('string');
 
     const databaseAdapter2 = new StubPostgresDatabaseAdapter<NumberKeyFields, 'id'>(
       numberKeyEntityConfiguration,
       new Map(),
     );
-    const insertedObject2 = await databaseAdapter2.insertAsync(queryContext, {});
-    expect(typeof insertedObject2.id).toBe('number');
+    const insertedObjects2 = await databaseAdapter2.insertManyAsync(queryContext, [{}]);
+    expect(typeof insertedObjects2[0]!.id).toBe('number');
 
     const databaseAdapter3 = new StubPostgresDatabaseAdapter<DateIDTestFields, 'id'>(
       dateIDTestEntityConfiguration,
       new Map(),
     );
-    await expect(databaseAdapter3.insertAsync(queryContext, {})).rejects.toThrow(
+    await expect(databaseAdapter3.insertManyAsync(queryContext, [{}])).rejects.toThrow(
       'Unsupported ID type for StubPostgresDatabaseAdapter: DateField',
     );
   });

packages/entity-database-adapter-knex/src/PostgresEntityDatabaseAdapter.ts

@@ -277,37 +277,97 @@ export class PostgresEntityDatabaseAdapter<
     return parseInt(String(result[0][RESERVED_ENTITY_COUNT_QUERY_ALIAS]), 10);
   }
 
-  protected async insertInternalAsync(
+  protected async insertManyInternalAsync(
     queryInterface: Knex,
     tableName: string,
-    object: object,
+    objects: readonly object[],
   ): Promise<object[]> {
     return await wrapNativePostgresCallAsync(() =>
-      queryInterface.insert(object).into(tableName).returning('*'),
+      queryInterface
+        .insert([...objects])
+        .into(tableName)
+        .returning('*'),
     );
   }
 
-  protected async updateInternalAsync(
+  protected async updateManyInternalAsync(
     queryInterface: Knex,
     tableName: string,
     tableIdField: string,
-    id: any,
-    object: object,
-  ): Promise<{ updatedRowCount: number }> {
-    const updatedRowCount = await wrapNativePostgresCallAsync(() =>
-      queryInterface.update(object).into(tableName).where(tableIdField, id),
-    );
-    return { updatedRowCount };
+    items: readonly { id: any; object: object }[],
+  ): Promise<readonly { updatedRowCount: number }[]> {
+    if (items.length === 0) {
+      return [];
+    }
+
+    if (items.length === 1) {
+      const item = items[0]!;
+      const updatedRowCount = await wrapNativePostgresCallAsync(() =>
+        queryInterface.update(item.object).into(tableName).where(tableIdField, item.id),
+      );
+      return [{ updatedRowCount }];
+    }
+
+    // Bulk update using UPDATE ... FROM (VALUES ...) for same-column-set items.
+    // All items are guaranteed to have the same set of columns.
+    const columns = Object.keys(items[0]!.object);
+    const allColumns = [tableIdField, ...columns];
+
+    const valuePlaceholders = items
+      .map(() => `(${allColumns.map(() => '?').join(', ')})`)
+      .join(', ');
+    const bindings: any[] = items.flatMap((item) => [
+      item.id,
+      ...columns.map((col) => (item.object as Record<string, any>)[col]),
+    ]);
+
+    const setClause = columns.map(() => `?? = ??`).join(', ');
+    const setBindings = columns.flatMap((col) => [col, `_data_table_.${col}`]);
+
+    const columnList = allColumns.map(() => '??').join(', ');
+    const columnBindings = allColumns;
+
+    const sql = [
+      `UPDATE ?? SET ${setClause}`,
+      `FROM (VALUES ${valuePlaceholders}) AS "_data_table_"(${columnList})`,
+      `WHERE ?? = ??`,
+      `RETURNING ??`,
+    ].join(' ');
+
+    const allBindings = [
+      tableName,
+      ...setBindings,
+      ...bindings,
+      ...columnBindings,
+      `${tableName}.${tableIdField}`,
+      `_data_table_.${tableIdField}`,
+      `${tableName}.${tableIdField}`,
+    ];
+
+    const result = await wrapNativePostgresCallAsync(() => queryInterface.raw(sql, allBindings));
+
+    const updatedIdCounts = new Map<any, number>();
+    for (const row of result.rows as Record<string, any>[]) {
+      const id = row[tableIdField];
+      updatedIdCounts.set(id, (updatedIdCounts.get(id) ?? 0) + 1);
+    }
+
+    return items.map((item) => ({
+      updatedRowCount: updatedIdCounts.get(item.id) ?? 0,
+    }));
   }
 
-  protected async deleteInternalAsync(
+  protected async deleteManyInternalAsync(
     queryInterface: Knex,
     tableName: string,
     tableIdField: string,
-    id: any,
+    ids: readonly any[],
   ): Promise<number> {
     return await wrapNativePostgresCallAsync(() =>
-      queryInterface.into(tableName).where(tableIdField, id).del(),
+      queryInterface
+        .from(tableName)
+        .whereIn(tableIdField, [...ids])
+        .del(),
     );
   }
 }