Function ProcessBatchUpdateCore

pg_documentdb/src/commands/update.c:1492–1595

static void
ProcessBatchUpdateCore(MongoCollection *collection, List *updates, text *transactionId,
					   BatchUpdateResult *batchResult, bool isOrdered,
					   bool forceInlineWrites, ExprEvalState *stateForSchemaValidation,
					   WriteMode writeMode)
{
	batchResult->ok = 1;
	batchResult->rowsMatched = 0;
	batchResult->rowsModified = 0;
	batchResult->writeErrors = NIL;
	batchResult->upserted = NIL;

	text *subTransactionId = transactionId;

	if (list_length(updates) == 1)
	{
		int updateIndex = 0;
		if (writeMode == WriteMode_Txn_Proc && transactionId == NULL)
		{
			DoSingleUpdate(collection, linitial(updates), subTransactionId,
						   batchResult, updateIndex, forceInlineWrites,
						   stateForSchemaValidation);
			return;
		}

		ereport(DEBUG1, (errmsg("Using single update with subtransaction")));
		DoSingleUpdateWithSubTxn(collection, linitial(updates), subTransactionId,
								 batchResult, updateIndex, forceInlineWrites,
								 stateForSchemaValidation);
		return;
	}
	else
	{
		/*
		 * We cannot pass the same transactionId to ProcessUpdate when there are
		 * multiple updates, since they would be considered retries of each
		 * other. We pass NULL for now to disable retryable writes.
		 */
		subTransactionId = NULL;
	}

	int updateIndex = 0;
	int nextBatchAttemptIndex = -1;
	bool hasBatchUpdateFailed = false;

	ListCell *updateCell = NULL;
	while (updateIndex < list_length(updates))
	{
		CHECK_FOR_INTERRUPTS();
		if (writeMode == WriteMode_Bulk_Proc && updateIndex > 0)
		{
			/* For each iteration of the loop, commit prior work */
			bool setSnapshot = false;
			Oid shardTableOid = InvalidOid;
			CommitWriteProcedureAndReacquireCollectionLock(collection, shardTableOid,
														   setSnapshot);
		}