Function ProcessBatchInsertion

pg_documentdb/src/commands/insert.c:838–895

* ProcessBatchInsertion iterates over the inserts array and executes each * insertion in a subtransaction, to allow us to continue after an error. * * If batchSpec->isOrdered is false, we continue with remaining tasks and * track the error for the response. * * Using subtransactions is slightly different from Mongo, which effectively * does each insert operation in a separate transaction, b

 * the same overall UX.
 */
static void
ProcessBatchInsertion(MongoCollection *collection, BatchInsertionSpec *batchSpec,
					  text *transactionId, BatchInsertionResult *batchResult,
					  WriteMode writeMode)
{
	batchResult->ok = 1;
	batchResult->rowsInserted = 0;
	batchResult->writeErrors = NIL;

	ExprEvalState *evalState = NULL;

	/* Document validation occurs regardless of whether the validation action is set to error or warn.
	 * If validation fails and the action is error, an error is thrown; if the action is warn, a warning is logged.
	 * Since we do not need to log a warning in this context, we will avoid calling ValidateSchemaOnDocumentInsert when the validation action is set to warn.
	 */
	if (CheckSchemaValidationEnabled(collection, batchSpec->bypassDocumentValidation))
	{
		evalState = PrepareForSchemaValidation(collection->schemaValidator.validator,
											   batchResult->resultMemoryContext);
	}

	/*
	 * We cannot pass the same transactionId to BatchInsert when there are
	 * multiple inserts, since they would be considered retries of each
	 * other. We pass NULL for batch insert to disable retryable writes.
	 */
	if (list_length(batchSpec->documents) == 1)
	{
		/* So at this point, we have a single document */
		int insertIndex = 0;
		if (writeMode == WriteMode_Txn_Proc && transactionId == NULL)
		{
			DoSingleInsert(collection, batchSpec->insertShardOid,
						   linitial(batchSpec->documents), transactionId,
						   batchResult, insertIndex, evalState);
		}
		else
		{
			ereport(DEBUG1, (errmsg("Using single insert with subtransaction")));
			DoSingleInsertWithSubTxn(collection, batchSpec->insertShardOid,
									 linitial(batchSpec->documents), transactionId,
									 batchResult, insertIndex, evalState);
		}
	}
	else
	{
		/* The else scenario - we have no transactionId (or we ignore it)
		 * and/or we have more than 1 document. Do a batch insert directly.
		 */
		DoBatchInsertNoTransactionId(collection, batchSpec, batchResult, evalState,
									 writeMode);
	}

	if (evalState != NULL)
	{
		FreeExprEvalState(evalState, batchResult->resultMemoryContext);
	}
}