Skip to content

Commit bb1a2bb

Browse files
committed
feat(NODE-6337): implement client bulk write batching
1 parent ddcae44 commit bb1a2bb

File tree

3 files changed

+262
-152
lines changed

3 files changed

+262
-152
lines changed

src/operations/client_bulk_write/executor.ts

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -70,10 +70,14 @@ async function executeAcknowledged(
7070
): Promise<ClientBulkWriteResult> {
7171
const resultsMerger = new ClientBulkWriteResultsMerger(options);
7272
// For each command will will create and exhaust a cursor for the results.
73+
let currentBatchOffset = 0;
7374
for (const command of commands) {
7475
const cursor = new ClientBulkWriteCursor(client, command, options);
7576
const docs = await cursor.toArray();
76-
resultsMerger.merge(command.ops.documents, cursor.response, docs);
77+
const operations = command.ops.documents;
78+
resultsMerger.merge(currentBatchOffset, operations, cursor.response, docs);
79+
// Set the new batch index so we can back back to the index in the original models.
80+
currentBatchOffset += operations.length;
7781
}
7882
return resultsMerger.result;
7983
}

src/operations/client_bulk_write/results_merger.ts

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -44,11 +44,13 @@ export class ClientBulkWriteResultsMerger {
4444

4545
/**
4646
* Merge the results in the cursor to the existing result.
47+
* @param currentBatchOffset - The offset index to the original models.
4748
* @param response - The cursor response.
4849
* @param documents - The documents in the cursor.
4950
* @returns The current result.
5051
*/
5152
merge(
53+
currentBatchOffset: number,
5254
operations: Document[],
5355
response: ClientBulkWriteCursorResponse,
5456
documents: Document[]
@@ -69,7 +71,9 @@ export class ClientBulkWriteResultsMerger {
6971
const operation = operations[document.idx];
7072
// Handle insert results.
7173
if ('insert' in operation) {
72-
this.result.insertResults?.set(document.idx, { insertedId: operation.document._id });
74+
this.result.insertResults?.set(document.idx + currentBatchOffset, {
75+
insertedId: operation.document._id
76+
});
7377
}
7478
// Handle update results.
7579
if ('update' in operation) {
@@ -80,11 +84,13 @@ export class ClientBulkWriteResultsMerger {
8084
if (document.upserted) {
8185
result.upsertedId = document.upserted._id;
8286
}
83-
this.result.updateResults?.set(document.idx, result);
87+
this.result.updateResults?.set(document.idx + currentBatchOffset, result);
8488
}
8589
// Handle delete results.
8690
if ('delete' in operation) {
87-
this.result.deleteResults?.set(document.idx, { deletedCount: document.n });
91+
this.result.deleteResults?.set(document.idx + currentBatchOffset, {
92+
deletedCount: document.n
93+
});
8894
}
8995
}
9096
}

0 commit comments

Comments
 (0)