opusdev/vector-similarity-api
1
1import { type BSONSerializeOptions, type Document, EJSON, resolveBSONOptions } from '../bson';2import type { Collection } from '../collection';3import {4 type AnyError,5 MongoBatchReExecutionError,6 MONGODB_ERROR_CODES,7 MongoInvalidArgumentError,8 MongoRuntimeError,9 MongoServerError,10 MongoWriteConcernError11} from '../error';12import type { Filter, OneOrMore, OptionalId, UpdateFilter, WithoutId } from '../mongo_types';13import type { CollationOptions, CommandOperationOptions } from '../operations/command';14import { DeleteOperation, type DeleteStatement, makeDeleteStatement } from '../operations/delete';15import { executeOperation } from '../operations/execute_operation';16import { InsertOperation } from '../operations/insert';17import { type Hint } from '../operations/operation';18import { makeUpdateStatement, UpdateOperation, type UpdateStatement } from '../operations/update';19import type { Topology } from '../sdam/topology';20import { type Sort } from '../sort';21import { TimeoutContext } from '../timeout';22import {23 applyRetryableWrites,24 getTopology,25 hasAtomicOperators,26 maybeAddIdToDocuments,27 type MongoDBNamespace,28 resolveOptions29} from '../utils';30import { WriteConcern } from '../write_concern';31 32/** @public */33export const BatchType = Object.freeze({34 INSERT: 1,35 UPDATE: 2,36 DELETE: 337} as const);38 39/** @public */40export type BatchType = (typeof BatchType)[keyof typeof BatchType];41 42/** @public */43export interface InsertOneModel<TSchema extends Document = Document> {44 /** The document to insert. */45 document: OptionalId<TSchema>;46}47 48/** @public */49export interface DeleteOneModel<TSchema extends Document = Document> {50 /** The filter to limit the deleted documents. */51 filter: Filter<TSchema>;52 /** Specifies a collation. */53 collation?: CollationOptions;54 /** The index to use. If specified, then the query system will only consider plans using the hinted index. */55 hint?: Hint;56}57 58/** @public */59export interface DeleteManyModel<TSchema extends Document = Document> {60 /** The filter to limit the deleted documents. */61 filter: Filter<TSchema>;62 /** Specifies a collation. */63 collation?: CollationOptions;64 /** The index to use. If specified, then the query system will only consider plans using the hinted index. */65 hint?: Hint;66}67 68/** @public */69export interface ReplaceOneModel<TSchema extends Document = Document> {70 /** The filter that specifies which document to replace. In the case of multiple matches, the first document matched is replaced. */71 filter: Filter<TSchema>;72 /** The document with which to replace the matched document. */73 replacement: WithoutId<TSchema>;74 /** Specifies a collation. */75 collation?: CollationOptions;76 /** The index to use. If specified, then the query system will only consider plans using the hinted index. */77 hint?: Hint;78 /** When true, creates a new document if no document matches the query. */79 upsert?: boolean;80 /** Specifies the sort order for the documents matched by the filter. */81 sort?: Sort;82}83 84/** @public */85export interface UpdateOneModel<TSchema extends Document = Document> {86 /** The filter that specifies which document to update. In the case of multiple matches, the first document matched is updated. */87 filter: Filter<TSchema>;88 /**89 * The modifications to apply. The value can be either:90 * UpdateFilter<TSchema> - A document that contains update operator expressions,91 * Document[] - an aggregation pipeline.92 */93 update: UpdateFilter<TSchema> | Document[];94 /** A set of filters specifying to which array elements an update should apply. */95 arrayFilters?: Document[];96 /** Specifies a collation. */97 collation?: CollationOptions;98 /** The index to use. If specified, then the query system will only consider plans using the hinted index. */99 hint?: Hint;100 /** When true, creates a new document if no document matches the query. */101 upsert?: boolean;102 /** Specifies the sort order for the documents matched by the filter. */103 sort?: Sort;104}105 106/** @public */107export interface UpdateManyModel<TSchema extends Document = Document> {108 /** The filter to limit the updated documents. */109 filter: Filter<TSchema>;110 /**111 * The modifications to apply. The value can be either:112 * UpdateFilter<TSchema> - A document that contains update operator expressions,113 * Document[] - an aggregation pipeline.114 */115 update: UpdateFilter<TSchema> | Document[];116 /** A set of filters specifying to which array elements an update should apply. */117 arrayFilters?: Document[];118 /** Specifies a collation. */119 collation?: CollationOptions;120 /** The index to use. If specified, then the query system will only consider plans using the hinted index. */121 hint?: Hint;122 /** When true, creates a new document if no document matches the query. */123 upsert?: boolean;124}125 126/** @public */127export type AnyBulkWriteOperation<TSchema extends Document = Document> =128 | { insertOne: InsertOneModel<TSchema> }129 | { replaceOne: ReplaceOneModel<TSchema> }130 | { updateOne: UpdateOneModel<TSchema> }131 | { updateMany: UpdateManyModel<TSchema> }132 | { deleteOne: DeleteOneModel<TSchema> }133 | { deleteMany: DeleteManyModel<TSchema> };134 135/** @internal */136export interface BulkResult {137 ok: number;138 writeErrors: WriteError[];139 writeConcernErrors: WriteConcernError[];140 insertedIds: Document[];141 nInserted: number;142 nUpserted: number;143 nMatched: number;144 nModified: number;145 nRemoved: number;146 upserted: Document[];147}148 149/**150 * Keeps the state of a unordered batch so we can rewrite the results151 * correctly after command execution152 *153 * @public154 */155export class Batch<T = Document> {156 originalZeroIndex: number;157 currentIndex: number;158 originalIndexes: number[];159 batchType: BatchType;160 operations: T[];161 size: number;162 sizeBytes: number;163 164 constructor(batchType: BatchType, originalZeroIndex: number) {165 this.originalZeroIndex = originalZeroIndex;166 this.currentIndex = 0;167 this.originalIndexes = [];168 this.batchType = batchType;169 this.operations = [];170 this.size = 0;171 this.sizeBytes = 0;172 }173}174 175/**176 * @public177 * The result of a bulk write.178 */179export class BulkWriteResult {180 private readonly result: BulkResult;181 /** Number of documents inserted. */182 readonly insertedCount: number;183 /** Number of documents matched for update. */184 readonly matchedCount: number;185 /** Number of documents modified. */186 readonly modifiedCount: number;187 /** Number of documents deleted. */188 readonly deletedCount: number;189 /** Number of documents upserted. */190 readonly upsertedCount: number;191 /** Upserted document generated Id's, hash key is the index of the originating operation */192 readonly upsertedIds: { [key: number]: any };193 /** Inserted document generated Id's, hash key is the index of the originating operation */194 readonly insertedIds: { [key: number]: any };195 196 private static generateIdMap(ids: Document[]): { [key: number]: any } {197 const idMap: { [index: number]: any } = {};198 for (const doc of ids) {199 idMap[doc.index] = doc._id;200 }201 return idMap;202 }203 204 /**205 * Create a new BulkWriteResult instance206 * @internal207 */208 constructor(bulkResult: BulkResult, isOrdered: boolean) {209 this.result = bulkResult;210 this.insertedCount = this.result.nInserted ?? 0;211 this.matchedCount = this.result.nMatched ?? 0;212 this.modifiedCount = this.result.nModified ?? 0;213 this.deletedCount = this.result.nRemoved ?? 0;214 this.upsertedCount = this.result.upserted.length ?? 0;215 this.upsertedIds = BulkWriteResult.generateIdMap(this.result.upserted);216 this.insertedIds = BulkWriteResult.generateIdMap(217 this.getSuccessfullyInsertedIds(bulkResult, isOrdered)218 );219 Object.defineProperty(this, 'result', { value: this.result, enumerable: false });220 }221 222 /** Evaluates to true if the bulk operation correctly executes */223 get ok(): number {224 return this.result.ok;225 }226 227 /**228 * Returns document_ids that were actually inserted229 * @internal230 */231 private getSuccessfullyInsertedIds(bulkResult: BulkResult, isOrdered: boolean): Document[] {232 if (bulkResult.writeErrors.length === 0) return bulkResult.insertedIds;233 234 if (isOrdered) {235 return bulkResult.insertedIds.slice(0, bulkResult.writeErrors[0].index);236 }237 238 return bulkResult.insertedIds.filter(239 ({ index }) => !bulkResult.writeErrors.some(writeError => index === writeError.index)240 );241 }242 243 /** Returns the upserted id at the given index */244 getUpsertedIdAt(index: number): Document | undefined {245 return this.result.upserted[index];246 }247 248 /** Returns raw internal result */249 getRawResponse(): Document {250 return this.result;251 }252 253 /** Returns true if the bulk operation contains a write error */254 hasWriteErrors(): boolean {255 return this.result.writeErrors.length > 0;256 }257 258 /** Returns the number of write errors from the bulk operation */259 getWriteErrorCount(): number {260 return this.result.writeErrors.length;261 }262 263 /** Returns a specific write error object */264 getWriteErrorAt(index: number): WriteError | undefined {265 return index < this.result.writeErrors.length ? this.result.writeErrors[index] : undefined;266 }267 268 /** Retrieve all write errors */269 getWriteErrors(): WriteError[] {270 return this.result.writeErrors;271 }272 273 /** Retrieve the write concern error if one exists */274 getWriteConcernError(): WriteConcernError | undefined {275 if (this.result.writeConcernErrors.length === 0) {276 return;277 } else if (this.result.writeConcernErrors.length === 1) {278 // Return the error279 return this.result.writeConcernErrors[0];280 } else {281 // Combine the errors282 let errmsg = '';283 for (let i = 0; i < this.result.writeConcernErrors.length; i++) {284 const err = this.result.writeConcernErrors[i];285 errmsg = errmsg + err.errmsg;286 287 // TODO: Something better288 if (i === 0) errmsg = errmsg + ' and ';289 }290 291 return new WriteConcernError({ errmsg, code: MONGODB_ERROR_CODES.WriteConcernTimeout });292 }293 }294 295 toString(): string {296 return `BulkWriteResult(${EJSON.stringify(this.result)})`;297 }298 299 isOk(): boolean {300 return this.result.ok === 1;301 }302}303 304/** @public */305export interface WriteConcernErrorData {306 code: number;307 errmsg: string;308 errInfo?: Document;309}310 311/**312 * An error representing a failure by the server to apply the requested write concern to the bulk operation.313 * @public314 * @category Error315 */316export class WriteConcernError {317 /** @internal */318 private serverError: WriteConcernErrorData;319 320 constructor(error: WriteConcernErrorData) {321 this.serverError = error;322 }323 324 /** Write concern error code. */325 get code(): number | undefined {326 return this.serverError.code;327 }328 329 /** Write concern error message. */330 get errmsg(): string | undefined {331 return this.serverError.errmsg;332 }333 334 /** Write concern error info. */335 get errInfo(): Document | undefined {336 return this.serverError.errInfo;337 }338 339 toJSON(): WriteConcernErrorData {340 return this.serverError;341 }342 343 toString(): string {344 return `WriteConcernError(${this.errmsg})`;345 }346}347 348/** @public */349export interface BulkWriteOperationError {350 index: number;351 code: number;352 errmsg: string;353 errInfo: Document;354 op: Document | UpdateStatement | DeleteStatement;355}356 357/**358 * An error that occurred during a BulkWrite on the server.359 * @public360 * @category Error361 */362export class WriteError {363 err: BulkWriteOperationError;364 365 constructor(err: BulkWriteOperationError) {366 this.err = err;367 }368 369 /** WriteError code. */370 get code(): number {371 return this.err.code;372 }373 374 /** WriteError original bulk operation index. */375 get index(): number {376 return this.err.index;377 }378 379 /** WriteError message. */380 get errmsg(): string | undefined {381 return this.err.errmsg;382 }383 384 /** WriteError details. */385 get errInfo(): Document | undefined {386 return this.err.errInfo;387 }388 389 /** Returns the underlying operation that caused the error */390 getOperation(): Document {391 return this.err.op;392 }393 394 toJSON(): { code: number; index: number; errmsg?: string; op: Document } {395 return { code: this.err.code, index: this.err.index, errmsg: this.err.errmsg, op: this.err.op };396 }397 398 toString(): string {399 return `WriteError(${JSON.stringify(this.toJSON())})`;400 }401}402 403/** Merges results into shared data structure */404export function mergeBatchResults(405 batch: Batch,406 bulkResult: BulkResult,407 err?: AnyError,408 result?: Document409): void {410 // If we have an error set the result to be the err object411 if (err) {412 result = err;413 } else if (result && result.result) {414 result = result.result;415 }416 417 if (result == null) {418 return;419 }420 421 // Do we have a top level error stop processing and return422 if (result.ok === 0 && bulkResult.ok === 1) {423 bulkResult.ok = 0;424 425 const writeError = {426 index: 0,427 code: result.code || 0,428 errmsg: result.message,429 errInfo: result.errInfo,430 op: batch.operations[0]431 };432 433 bulkResult.writeErrors.push(new WriteError(writeError));434 return;435 } else if (result.ok === 0 && bulkResult.ok === 0) {436 return;437 }438 439 // If we have an insert Batch type440 if (isInsertBatch(batch) && result.n) {441 bulkResult.nInserted = bulkResult.nInserted + result.n;442 }443 444 // If we have an insert Batch type445 if (isDeleteBatch(batch) && result.n) {446 bulkResult.nRemoved = bulkResult.nRemoved + result.n;447 }448 449 let nUpserted = 0;450 451 // We have an array of upserted values, we need to rewrite the indexes452 if (Array.isArray(result.upserted)) {453 nUpserted = result.upserted.length;454 455 for (let i = 0; i < result.upserted.length; i++) {456 bulkResult.upserted.push({457 index: result.upserted[i].index + batch.originalZeroIndex,458 _id: result.upserted[i]._id459 });460 }461 } else if (result.upserted) {462 nUpserted = 1;463 464 bulkResult.upserted.push({465 index: batch.originalZeroIndex,466 _id: result.upserted467 });468 }469 470 // If we have an update Batch type471 if (isUpdateBatch(batch) && result.n) {472 const nModified = result.nModified;473 bulkResult.nUpserted = bulkResult.nUpserted + nUpserted;474 bulkResult.nMatched = bulkResult.nMatched + (result.n - nUpserted);475 476 if (typeof nModified === 'number') {477 bulkResult.nModified = bulkResult.nModified + nModified;478 } else {479 bulkResult.nModified = 0;480 }481 }482 483 if (Array.isArray(result.writeErrors)) {484 for (let i = 0; i < result.writeErrors.length; i++) {485 const writeError = {486 index: batch.originalIndexes[result.writeErrors[i].index],487 code: result.writeErrors[i].code,488 errmsg: result.writeErrors[i].errmsg,489 errInfo: result.writeErrors[i].errInfo,490 op: batch.operations[result.writeErrors[i].index]491 };492 493 bulkResult.writeErrors.push(new WriteError(writeError));494 }495 }496 497 if (result.writeConcernError) {498 bulkResult.writeConcernErrors.push(new WriteConcernError(result.writeConcernError));499 }500}501 502async function executeCommands(503 bulkOperation: BulkOperationBase,504 options: BulkWriteOptions & { timeoutContext?: TimeoutContext | null }505): Promise<BulkWriteResult> {506 if (bulkOperation.s.batches.length === 0) {507 return new BulkWriteResult(bulkOperation.s.bulkResult, bulkOperation.isOrdered);508 }509 510 for (const batch of bulkOperation.s.batches) {511 const finalOptions = resolveOptions(bulkOperation, {512 ...options,513 ordered: bulkOperation.isOrdered514 });515 516 if (finalOptions.bypassDocumentValidation !== true) {517 delete finalOptions.bypassDocumentValidation;518 }519 520 // Is the bypassDocumentValidation options specific521 if (bulkOperation.s.bypassDocumentValidation === true) {522 finalOptions.bypassDocumentValidation = true;523 }524 525 // Is the checkKeys option disabled526 if (bulkOperation.s.checkKeys === false) {527 finalOptions.checkKeys = false;528 }529 530 if (finalOptions.retryWrites) {531 if (isUpdateBatch(batch)) {532 finalOptions.retryWrites =533 finalOptions.retryWrites && !batch.operations.some(op => op.multi);534 }535 536 if (isDeleteBatch(batch)) {537 finalOptions.retryWrites =538 finalOptions.retryWrites && !batch.operations.some(op => op.limit === 0);539 }540 }541 542 const operation = isInsertBatch(batch)543 ? new InsertOperation(bulkOperation.s.namespace, batch.operations, finalOptions)544 : isUpdateBatch(batch)545 ? new UpdateOperation(bulkOperation.s.namespace, batch.operations, finalOptions)546 : isDeleteBatch(batch)547 ? new DeleteOperation(bulkOperation.s.namespace, batch.operations, finalOptions)548 : null;549 550 if (operation == null) throw new MongoRuntimeError(`Unknown batchType: ${batch.batchType}`);551 552 let thrownError = null;553 let result;554 try {555 result = await executeOperation(556 bulkOperation.s.collection.client,557 operation,558 finalOptions.timeoutContext559 );560 } catch (error) {561 thrownError = error;562 }563 564 if (thrownError != null) {565 if (thrownError instanceof MongoWriteConcernError) {566 mergeBatchResults(batch, bulkOperation.s.bulkResult, thrownError, result);567 const writeResult = new BulkWriteResult(568 bulkOperation.s.bulkResult,569 bulkOperation.isOrdered570 );571 572 throw new MongoBulkWriteError(573 {574 message: thrownError.result.writeConcernError.errmsg,575 code: thrownError.result.writeConcernError.code576 },577 writeResult578 );579 } else {580 // Error is a driver related error not a bulk op error, return early581 throw new MongoBulkWriteError(582 thrownError,583 new BulkWriteResult(bulkOperation.s.bulkResult, bulkOperation.isOrdered)584 );585 }586 }587 588 mergeBatchResults(batch, bulkOperation.s.bulkResult, thrownError, result);589 const writeResult = new BulkWriteResult(bulkOperation.s.bulkResult, bulkOperation.isOrdered);590 bulkOperation.handleWriteError(writeResult);591 }592 593 bulkOperation.s.batches.length = 0;594 595 const writeResult = new BulkWriteResult(bulkOperation.s.bulkResult, bulkOperation.isOrdered);596 bulkOperation.handleWriteError(writeResult);597 return writeResult;598}599 600/**601 * An error indicating an unsuccessful Bulk Write602 * @public603 * @category Error604 */605export class MongoBulkWriteError extends MongoServerError {606 result: BulkWriteResult;607 writeErrors: OneOrMore<WriteError> = [];608 err?: WriteConcernError;609 610 /**611 * **Do not use this constructor!**612 *613 * Meant for internal use only.614 *615 * @remarks616 * This class is only meant to be constructed within the driver. This constructor is617 * not subject to semantic versioning compatibility guarantees and may change at any time.618 *619 * @public620 **/621 constructor(622 error:623 | { message: string; code: number; writeErrors?: WriteError[] }624 | WriteConcernError625 | AnyError,626 result: BulkWriteResult627 ) {628 super(error);629 630 if (error instanceof WriteConcernError) this.err = error;631 else if (!(error instanceof Error)) {632 this.message = error.message;633 this.code = error.code;634 this.writeErrors = error.writeErrors ?? [];635 }636 637 this.result = result;638 Object.assign(this, error);639 }640 641 override get name(): string {642 return 'MongoBulkWriteError';643 }644 645 /** Number of documents inserted. */646 get insertedCount(): number {647 return this.result.insertedCount;648 }649 /** Number of documents matched for update. */650 get matchedCount(): number {651 return this.result.matchedCount;652 }653 /** Number of documents modified. */654 get modifiedCount(): number {655 return this.result.modifiedCount;656 }657 /** Number of documents deleted. */658 get deletedCount(): number {659 return this.result.deletedCount;660 }661 /** Number of documents upserted. */662 get upsertedCount(): number {663 return this.result.upsertedCount;664 }665 /** Inserted document generated Id's, hash key is the index of the originating operation */666 get insertedIds(): { [key: number]: any } {667 return this.result.insertedIds;668 }669 /** Upserted document generated Id's, hash key is the index of the originating operation */670 get upsertedIds(): { [key: number]: any } {671 return this.result.upsertedIds;672 }673}674 675/**676 * A builder object that is returned from {@link BulkOperationBase#find}.677 * Is used to build a write operation that involves a query filter.678 *679 * @public680 */681export class FindOperators {682 bulkOperation: BulkOperationBase;683 684 /**685 * Creates a new FindOperators object.686 * @internal687 */688 constructor(bulkOperation: BulkOperationBase) {689 this.bulkOperation = bulkOperation;690 }691 692 /** Add a multiple update operation to the bulk operation */693 update(updateDocument: Document | Document[]): BulkOperationBase {694 const currentOp = buildCurrentOp(this.bulkOperation);695 return this.bulkOperation.addToOperationsList(696 BatchType.UPDATE,697 makeUpdateStatement(currentOp.selector, updateDocument, {698 ...currentOp,699 multi: true700 })701 );702 }703 704 /** Add a single update operation to the bulk operation */705 updateOne(updateDocument: Document | Document[]): BulkOperationBase {706 if (!hasAtomicOperators(updateDocument, this.bulkOperation.bsonOptions)) {707 throw new MongoInvalidArgumentError('Update document requires atomic operators');708 }709 710 const currentOp = buildCurrentOp(this.bulkOperation);711 return this.bulkOperation.addToOperationsList(712 BatchType.UPDATE,713 makeUpdateStatement(currentOp.selector, updateDocument, { ...currentOp, multi: false })714 );715 }716 717 /** Add a replace one operation to the bulk operation */718 replaceOne(replacement: Document): BulkOperationBase {719 if (hasAtomicOperators(replacement)) {720 throw new MongoInvalidArgumentError('Replacement document must not use atomic operators');721 }722 723 const currentOp = buildCurrentOp(this.bulkOperation);724 return this.bulkOperation.addToOperationsList(725 BatchType.UPDATE,726 makeUpdateStatement(currentOp.selector, replacement, { ...currentOp, multi: false })727 );728 }729 730 /** Add a delete one operation to the bulk operation */731 deleteOne(): BulkOperationBase {732 const currentOp = buildCurrentOp(this.bulkOperation);733 return this.bulkOperation.addToOperationsList(734 BatchType.DELETE,735 makeDeleteStatement(currentOp.selector, { ...currentOp, limit: 1 })736 );737 }738 739 /** Add a delete many operation to the bulk operation */740 delete(): BulkOperationBase {741 const currentOp = buildCurrentOp(this.bulkOperation);742 return this.bulkOperation.addToOperationsList(743 BatchType.DELETE,744 makeDeleteStatement(currentOp.selector, { ...currentOp, limit: 0 })745 );746 }747 748 /** Upsert modifier for update bulk operation, noting that this operation is an upsert. */749 upsert(): this {750 if (!this.bulkOperation.s.currentOp) {751 this.bulkOperation.s.currentOp = {};752 }753 754 this.bulkOperation.s.currentOp.upsert = true;755 return this;756 }757 758 /** Specifies the collation for the query condition. */759 collation(collation: CollationOptions): this {760 if (!this.bulkOperation.s.currentOp) {761 this.bulkOperation.s.currentOp = {};762 }763 764 this.bulkOperation.s.currentOp.collation = collation;765 return this;766 }767 768 /** Specifies arrayFilters for UpdateOne or UpdateMany bulk operations. */769 arrayFilters(arrayFilters: Document[]): this {770 if (!this.bulkOperation.s.currentOp) {771 this.bulkOperation.s.currentOp = {};772 }773 774 this.bulkOperation.s.currentOp.arrayFilters = arrayFilters;775 return this;776 }777 778 /** Specifies hint for the bulk operation. */779 hint(hint: Hint): this {780 if (!this.bulkOperation.s.currentOp) {781 this.bulkOperation.s.currentOp = {};782 }783 784 this.bulkOperation.s.currentOp.hint = hint;785 return this;786 }787}788 789/** @internal */790export interface BulkOperationPrivate {791 bulkResult: BulkResult;792 currentBatch?: Batch;793 currentIndex: number;794 // ordered specific795 currentBatchSize: number;796 currentBatchSizeBytes: number;797 // unordered specific798 currentInsertBatch?: Batch;799 currentUpdateBatch?: Batch;800 currentRemoveBatch?: Batch;801 batches: Batch[];802 // Write concern803 writeConcern?: WriteConcern;804 // Max batch size options805 maxBsonObjectSize: number;806 maxBatchSizeBytes: number;807 maxWriteBatchSize: number;808 maxKeySize: number;809 // Namespace810 namespace: MongoDBNamespace;811 // Topology812 topology: Topology;813 // Options814 options: BulkWriteOptions;815 // BSON options816 bsonOptions: BSONSerializeOptions;817 // Document used to build a bulk operation818 currentOp?: Document;819 // Executed820 executed: boolean;821 // Collection822 collection: Collection;823 // Fundamental error824 err?: AnyError;825 // check keys826 checkKeys: boolean;827 bypassDocumentValidation?: boolean;828}829 830/** @public */831export interface BulkWriteOptions extends CommandOperationOptions {832 /**833 * Allow driver to bypass schema validation.834 * @defaultValue `false` - documents will be validated by default835 **/836 bypassDocumentValidation?: boolean;837 /**838 * If true, when an insert fails, don't execute the remaining writes.839 * If false, continue with remaining inserts when one fails.840 * @defaultValue `true` - inserts are ordered by default841 */842 ordered?: boolean;843 /**844 * Force server to assign _id values instead of driver.845 * @defaultValue `false` - the driver generates `_id` fields by default846 **/847 forceServerObjectId?: boolean;848 /** Map of parameter names and values that can be accessed using $$var (requires MongoDB 5.0). */849 let?: Document;850 851 /** @internal */852 timeoutContext?: TimeoutContext;853}854 855/** @public */856export abstract class BulkOperationBase {857 isOrdered: boolean;858 /** @internal */859 s: BulkOperationPrivate;860 operationId?: number;861 private collection: Collection;862 863 /**864 * Create a new OrderedBulkOperation or UnorderedBulkOperation instance865 * @internal866 */867 constructor(collection: Collection, options: BulkWriteOptions, isOrdered: boolean) {868 this.collection = collection;869 // determine whether bulkOperation is ordered or unordered870 this.isOrdered = isOrdered;871 872 const topology = getTopology(collection);873 options = options == null ? {} : options;874 // TODO Bring from driver information in hello875 // Get the namespace for the write operations876 const namespace = collection.s.namespace;877 // Used to mark operation as executed878 const executed = false;879 880 // Current item881 const currentOp = undefined;882 883 // Set max byte size884 const hello = topology.lastHello();885 886 // If we have autoEncryption on, batch-splitting must be done on 2mb chunks, but single documents887 // over 2mb are still allowed888 const usingAutoEncryption = !!(topology.s.options && topology.s.options.autoEncrypter);889 const maxBsonObjectSize =890 hello && hello.maxBsonObjectSize ? hello.maxBsonObjectSize : 1024 * 1024 * 16;891 const maxBatchSizeBytes = usingAutoEncryption ? 1024 * 1024 * 2 : maxBsonObjectSize;892 const maxWriteBatchSize = hello && hello.maxWriteBatchSize ? hello.maxWriteBatchSize : 1000;893 894 // Calculates the largest possible size of an Array key, represented as a BSON string895 // element. This calculation:896 // 1 byte for BSON type897 // # of bytes = length of (string representation of (maxWriteBatchSize - 1))898 // + 1 bytes for null terminator899 const maxKeySize = (maxWriteBatchSize - 1).toString(10).length + 2;900 901 // Final options for retryable writes902 let finalOptions = Object.assign({}, options);903 finalOptions = applyRetryableWrites(finalOptions, collection.db);904 905 // Final results906 const bulkResult: BulkResult = {907 ok: 1,908 writeErrors: [],909 writeConcernErrors: [],910 insertedIds: [],911 nInserted: 0,912 nUpserted: 0,913 nMatched: 0,914 nModified: 0,915 nRemoved: 0,916 upserted: []917 };918 919 // Internal state920 this.s = {921 // Final result922 bulkResult,923 // Current batch state924 currentBatch: undefined,925 currentIndex: 0,926 // ordered specific927 currentBatchSize: 0,928 currentBatchSizeBytes: 0,929 // unordered specific930 currentInsertBatch: undefined,931 currentUpdateBatch: undefined,932 currentRemoveBatch: undefined,933 batches: [],934 // Write concern935 writeConcern: WriteConcern.fromOptions(options),936 // Max batch size options937 maxBsonObjectSize,938 maxBatchSizeBytes,939 maxWriteBatchSize,940 maxKeySize,941 // Namespace942 namespace,943 // Topology944 topology,945 // Options946 options: finalOptions,947 // BSON options948 bsonOptions: resolveBSONOptions(options),949 // Current operation950 currentOp,951 // Executed952 executed,953 // Collection954 collection,955 // Fundamental error956 err: undefined,957 // check keys958 checkKeys: typeof options.checkKeys === 'boolean' ? options.checkKeys : false959 };960 961 // bypass Validation962 if (options.bypassDocumentValidation === true) {963 this.s.bypassDocumentValidation = true;964 }965 }966 967 /**968 * Add a single insert document to the bulk operation969 *970 * @example971 * ```ts972 * const bulkOp = collection.initializeOrderedBulkOp();973 *974 * // Adds three inserts to the bulkOp.975 * bulkOp976 * .insert({ a: 1 })977 * .insert({ b: 2 })978 * .insert({ c: 3 });979 * await bulkOp.execute();980 * ```981 */982 insert(document: Document): BulkOperationBase {983 maybeAddIdToDocuments(this.collection, document, {984 forceServerObjectId: this.shouldForceServerObjectId()985 });986 987 return this.addToOperationsList(BatchType.INSERT, document);988 }989 990 /**991 * Builds a find operation for an update/updateOne/delete/deleteOne/replaceOne.992 * Returns a builder object used to complete the definition of the operation.993 *994 * @example995 * ```ts996 * const bulkOp = collection.initializeOrderedBulkOp();997 *998 * // Add an updateOne to the bulkOp999 * bulkOp.find({ a: 1 }).updateOne({ $set: { b: 2 } });1000 *1001 * // Add an updateMany to the bulkOp1002 * bulkOp.find({ c: 3 }).update({ $set: { d: 4 } });1003 *1004 * // Add an upsert1005 * bulkOp.find({ e: 5 }).upsert().updateOne({ $set: { f: 6 } });1006 *1007 * // Add a deletion1008 * bulkOp.find({ g: 7 }).deleteOne();1009 *1010 * // Add a multi deletion1011 * bulkOp.find({ h: 8 }).delete();1012 *1013 * // Add a replaceOne1014 * bulkOp.find({ i: 9 }).replaceOne({writeConcern: { j: 10 }});1015 *1016 * // Update using a pipeline (requires Mongodb 4.2 or higher)1017 * bulk.find({ k: 11, y: { $exists: true }, z: { $exists: true } }).updateOne([1018 * { $set: { total: { $sum: [ '$y', '$z' ] } } }1019 * ]);1020 *1021 * // All of the ops will now be executed1022 * await bulkOp.execute();1023 * ```1024 */1025 find(selector: Document): FindOperators {1026 if (!selector) {1027 throw new MongoInvalidArgumentError('Bulk find operation must specify a selector');1028 }1029 1030 // Save a current selector1031 this.s.currentOp = {1032 selector: selector1033 };1034 1035 return new FindOperators(this);1036 }1037 1038 /** Specifies a raw operation to perform in the bulk write. */1039 raw(op: AnyBulkWriteOperation): this {1040 if (op == null || typeof op !== 'object') {1041 throw new MongoInvalidArgumentError('Operation must be an object with an operation key');1042 }1043 if ('insertOne' in op) {1044 const forceServerObjectId = this.shouldForceServerObjectId();1045 const document =1046 op.insertOne && op.insertOne.document == null1047 ? // TODO(NODE-6003): remove support for omitting the `documents` subdocument in bulk inserts1048 (op.insertOne as Document)1049 : op.insertOne.document;1050 1051 maybeAddIdToDocuments(this.collection, document, { forceServerObjectId });1052 1053 return this.addToOperationsList(BatchType.INSERT, document);1054 }1055 1056 if ('replaceOne' in op || 'updateOne' in op || 'updateMany' in op) {1057 if ('replaceOne' in op) {1058 if ('q' in op.replaceOne) {1059 throw new MongoInvalidArgumentError('Raw operations are not allowed');1060 }1061 const updateStatement = makeUpdateStatement(1062 op.replaceOne.filter,1063 op.replaceOne.replacement,1064 { ...op.replaceOne, multi: false }1065 );1066 if (hasAtomicOperators(updateStatement.u)) {1067 throw new MongoInvalidArgumentError('Replacement document must not use atomic operators');1068 }1069 return this.addToOperationsList(BatchType.UPDATE, updateStatement);1070 }1071 1072 if ('updateOne' in op) {1073 if ('q' in op.updateOne) {1074 throw new MongoInvalidArgumentError('Raw operations are not allowed');1075 }1076 const updateStatement = makeUpdateStatement(op.updateOne.filter, op.updateOne.update, {1077 ...op.updateOne,1078 multi: false1079 });1080 if (!hasAtomicOperators(updateStatement.u, this.bsonOptions)) {1081 throw new MongoInvalidArgumentError('Update document requires atomic operators');1082 }1083 return this.addToOperationsList(BatchType.UPDATE, updateStatement);1084 }1085 1086 if ('updateMany' in op) {1087 if ('q' in op.updateMany) {1088 throw new MongoInvalidArgumentError('Raw operations are not allowed');1089 }1090 const updateStatement = makeUpdateStatement(op.updateMany.filter, op.updateMany.update, {1091 ...op.updateMany,1092 multi: true1093 });1094 if (!hasAtomicOperators(updateStatement.u, this.bsonOptions)) {1095 throw new MongoInvalidArgumentError('Update document requires atomic operators');1096 }1097 return this.addToOperationsList(BatchType.UPDATE, updateStatement);1098 }1099 }1100 1101 if ('deleteOne' in op) {1102 if ('q' in op.deleteOne) {1103 throw new MongoInvalidArgumentError('Raw operations are not allowed');1104 }1105 return this.addToOperationsList(1106 BatchType.DELETE,1107 makeDeleteStatement(op.deleteOne.filter, { ...op.deleteOne, limit: 1 })1108 );1109 }1110 1111 if ('deleteMany' in op) {1112 if ('q' in op.deleteMany) {1113 throw new MongoInvalidArgumentError('Raw operations are not allowed');1114 }1115 return this.addToOperationsList(1116 BatchType.DELETE,1117 makeDeleteStatement(op.deleteMany.filter, { ...op.deleteMany, limit: 0 })1118 );1119 }1120 1121 // otherwise an unknown operation was provided1122 throw new MongoInvalidArgumentError(1123 'bulkWrite only supports insertOne, updateOne, updateMany, deleteOne, deleteMany'1124 );1125 }1126 1127 get length(): number {1128 return this.s.currentIndex;1129 }1130 1131 get bsonOptions(): BSONSerializeOptions {1132 return this.s.bsonOptions;1133 }1134 1135 get writeConcern(): WriteConcern | undefined {1136 return this.s.writeConcern;1137 }1138 1139 get batches(): Batch[] {1140 const batches = [...this.s.batches];1141 if (this.isOrdered) {1142 if (this.s.currentBatch) batches.push(this.s.currentBatch);1143 } else {1144 if (this.s.currentInsertBatch) batches.push(this.s.currentInsertBatch);1145 if (this.s.currentUpdateBatch) batches.push(this.s.currentUpdateBatch);1146 if (this.s.currentRemoveBatch) batches.push(this.s.currentRemoveBatch);1147 }1148 return batches;1149 }1150 1151 async execute(options: BulkWriteOptions = {}): Promise<BulkWriteResult> {1152 if (this.s.executed) {1153 throw new MongoBatchReExecutionError();1154 }1155 1156 const writeConcern = WriteConcern.fromOptions(options);1157 if (writeConcern) {1158 this.s.writeConcern = writeConcern;1159 }1160 1161 // If we have current batch1162 if (this.isOrdered) {1163 if (this.s.currentBatch) this.s.batches.push(this.s.currentBatch);1164 } else {1165 if (this.s.currentInsertBatch) this.s.batches.push(this.s.currentInsertBatch);1166 if (this.s.currentUpdateBatch) this.s.batches.push(this.s.currentUpdateBatch);1167 if (this.s.currentRemoveBatch) this.s.batches.push(this.s.currentRemoveBatch);1168 }1169 // If we have no operations in the bulk raise an error1170 if (this.s.batches.length === 0) {1171 throw new MongoInvalidArgumentError('Invalid BulkOperation, Batch cannot be empty');1172 }1173 1174 this.s.executed = true;1175 const finalOptions = resolveOptions(this.collection, { ...this.s.options, ...options });1176 1177 // if there is no timeoutContext provided, create a timeoutContext and use it for1178 // all batches in the bulk operation1179 finalOptions.timeoutContext ??= TimeoutContext.create({1180 session: finalOptions.session,1181 timeoutMS: finalOptions.timeoutMS,1182 serverSelectionTimeoutMS: this.collection.client.s.options.serverSelectionTimeoutMS,1183 waitQueueTimeoutMS: this.collection.client.s.options.waitQueueTimeoutMS1184 });1185 1186 if (finalOptions.session == null) {1187 // if there is not an explicit session provided to `execute()`, create1188 // an implicit session and use that for all batches in the bulk operation1189 return await this.collection.client.withSession({ explicit: false }, async session => {1190 return await executeCommands(this, { ...finalOptions, session });1191 });1192 }1193 1194 return await executeCommands(this, { ...finalOptions });1195 }1196 1197 /**1198 * Handles the write error before executing commands1199 * @internal1200 */