PHPC-2493: Manager and Server::executeBulkWriteCommand()

Author: jmikola

src/MongoDB/BulkWriteCommand.c

@@ -34,6 +34,34 @@
 
 zend_class_entry* php_phongo_bulkwritecommand_ce;
 
+/* Creates a mongoc_bulkwriteopts_t from internal options, which should be freed
+ * by the caller. */
+mongoc_bulkwriteopts_t* phongo_bwc_assemble_opts(php_phongo_bulkwritecommand_t* intern)
+{
+	mongoc_bulkwriteopts_t* opts = mongoc_bulkwriteopts_new();
+
+	if (intern->bypass != PHONGO_BULKWRITECOMMAND_BYPASS_UNSET) {
+		mongoc_bulkwriteopts_set_bypassdocumentvalidation(opts, intern->bypass);
+	}
+
+	if (intern->comment) {
+		mongoc_bulkwriteopts_set_comment(opts, intern->comment);
+	}
+
+	if (intern->let) {
+		mongoc_bulkwriteopts_set_let(opts, intern->let);
+	}
+
+	mongoc_bulkwriteopts_set_ordered(opts, intern->ordered);
+	mongoc_bulkwriteopts_set_verboseresults(opts, intern->verbose);
+
+	if (intern->write_concern) {
+		mongoc_bulkwriteopts_set_writeconcern(opts, intern->write_concern);
+	}
+
+	return opts;
+}
+
 // TODO: Make this a common utility function to share with BulkWrite.c
 /* Extracts the "_id" field of a BSON document into a return value. */
 static void phongo_bwc_extract_id(bson_t* doc, zval** return_value)
@@ -381,17 +409,6 @@ static PHP_METHOD(MongoDB_Driver_BulkWriteCommand, deleteOne)
 	mongoc_bulkwrite_deleteoneopts_destroy(opts);
 }
 
-static PHP_METHOD(MongoDB_Driver_BulkWriteCommand, execute)
-{
-	php_phongo_bulkwritecommand_t* intern;
-
-	intern = Z_BULKWRITECOMMAND_OBJ_P(getThis());
-
-	PHONGO_PARSE_PARAMETERS_NONE();
-
-	RETURN_LONG(intern->num_ops);
-}
-
 static PHP_METHOD(MongoDB_Driver_BulkWriteCommand, insertOne)
 {
 	php_phongo_bulkwritecommand_t* intern;

src/MongoDB/BulkWriteCommand.h

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2024-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef PHONGO_BULKWRITECOMMAND_H
+#define PHONGO_BULKWRITECOMMAND_H
+
+#include "mongoc/mongoc.h"
+
+#include "phongo_structs.h"
+
+mongoc_bulkwriteopts_t* phongo_bwc_assemble_opts(php_phongo_bulkwritecommand_t* intern);
+
+#endif /* PHONGO_BULKWRITECOMMAND_H */

src/MongoDB/BulkWriteCommand.stub.php

@@ -25,6 +25,4 @@ final public function replaceOne(string $namespace, array|object $filter, array|
     final public function updateOne(string $namespace, array|object $filter, array|object $update, ?array $options = null): void {}
 
     final public function updateMany(string $namespace, array|object $filter, array|object $update, ?array $options = null): void {}
-
-    final public function execute(Manager|Server $managerOrServer): mixed {}
 }

src/MongoDB/BulkWriteCommand_arginfo.h

@@ -595,6 +595,42 @@ static PHP_METHOD(MongoDB_Driver_Manager, executeBulkWrite)
 	}
 }
 
+/* Executes a BulkWriteCommand (i.e. bulkWrite command for MongoDB 8.0+) */
+static PHP_METHOD(MongoDB_Driver_Manager, executeBulkWriteCommand)
+{
+	php_phongo_manager_t* intern;
+	zval*                   zbwc;
+	php_phongo_bulkwritecommand_t* bwc;
+	zval*                   zoptions      = NULL;
+	uint32_t                server_id    = 0;
+	zval*                   zsession     = NULL;
+
+	PHONGO_PARSE_PARAMETERS_START(1, 2)
+	Z_PARAM_OBJECT_OF_CLASS(zbwc, php_phongo_bulkwritecommand_ce)
+	Z_PARAM_OPTIONAL
+	Z_PARAM_ZVAL_OR_NULL(zoptions)
+	PHONGO_PARSE_PARAMETERS_END();
+
+	intern = Z_MANAGER_OBJ_P(getThis());
+	bwc   = Z_BULKWRITECOMMAND_OBJ_P(zbwc);
+
+	if (!phongo_parse_session(zoptions, intern->client, NULL, &zsession)) {
+		/* Exception should already have been thrown */
+		return;
+	}
+
+	if (!php_phongo_manager_select_server(true, false, NULL, zsession, intern->client, &server_id)) {
+		/* Exception should already have been thrown */
+		return;
+	}
+
+	/* If the Server was created in a different process, reset the client so
+	 * that its session pool is cleared. */
+	PHONGO_RESET_CLIENT_IF_PID_DIFFERS(intern, intern);
+
+	phongo_execute_bulkwritecommand(getThis(), bwc, zoptions, server_id, return_value);
+}
+
 /* Returns the autoEncryption.encryptedFieldsMap driver option */
 static PHP_METHOD(MongoDB_Driver_Manager, getEncryptedFieldsMap)
 {

src/MongoDB/Manager.c

@@ -18,6 +18,8 @@ final public function createClientEncryption(array $options): ClientEncryption {
 
     final public function executeBulkWrite(string $namespace, BulkWrite $bulk, array|WriteConcern|null $options = null): WriteResult {}
 
+    final public function executeBulkWriteCommand(BulkWriteCommand $bulkWriteCommand, ?array $options = null): ?BulkWriteCommandResult {}
+
     final public function executeCommand(string $db, Command $command, array|ReadPreference|null $options = null): Cursor {}
 
     final public function executeQuery(string $namespace, Query $query, array|ReadPreference|null $options = null): Cursor {}

src/MongoDB/Manager.stub.php

@@ -215,6 +215,30 @@ static PHP_METHOD(MongoDB_Driver_Server, executeBulkWrite)
 	}
 }
 
+/* Executes a BulkWriteCommand (i.e. bulkWrite command for MongoDB 8.0+) */
+static PHP_METHOD(MongoDB_Driver_Server, executeBulkWriteCommand)
+{
+	php_phongo_server_t* intern;
+	zval*                   zbwc;
+	php_phongo_bulkwritecommand_t* bwc;
+	zval*                   zoptions      = NULL;
+
+	PHONGO_PARSE_PARAMETERS_START(1, 2)
+	Z_PARAM_OBJECT_OF_CLASS(zbwc, php_phongo_bulkwritecommand_ce)
+	Z_PARAM_OPTIONAL
+	Z_PARAM_ZVAL_OR_NULL(zoptions)
+	PHONGO_PARSE_PARAMETERS_END();
+
+	intern = Z_SERVER_OBJ_P(getThis());
+	bwc   = Z_BULKWRITECOMMAND_OBJ_P(zbwc);
+
+	/* If the Server was created in a different process, reset the client so
+	 * that its session pool is cleared. */
+	PHONGO_RESET_CLIENT_IF_PID_DIFFERS(intern, Z_MANAGER_OBJ_P(&intern->manager));
+
+	phongo_execute_bulkwritecommand(&intern->manager, bwc, zoptions, intern->server_id, return_value);
+}
+
 /* Returns the hostname for this Server */
 static PHP_METHOD(MongoDB_Driver_Server, getHost)
 {

src/MongoDB/Manager_arginfo.h

@@ -74,6 +74,8 @@ final private function __construct() {}
 
     final public function executeBulkWrite(string $namespace, BulkWrite $bulkWrite, array|WriteConcern|null $options = null): WriteResult {}
 
+    final public function executeBulkWriteCommand(BulkWriteCommand $bulkWriteCommand, ?array $options = null): ?BulkWriteCommandResult {}
+
     final public function executeCommand(string $db, Command $command, array|ReadPreference|null $options = null): Cursor {}
 
     final public function executeQuery(string $namespace, Query $query, array|ReadPreference|null $options = null): Cursor {}

src/MongoDB/Server.c

@@ -27,6 +27,8 @@
 #include "phongo_execute.h"
 #include "phongo_util.h"
 
+#include "MongoDB/BulkWriteCommand.h"
+#include "MongoDB/BulkWriteCommandResult.h"
 #include "MongoDB/Cursor.h"
 #include "MongoDB/ReadPreference.h"
 #include "MongoDB/Session.h"
@@ -333,6 +335,83 @@ bool phongo_execute_bulk_write(zval* manager, const char* namespace, php_phongo_
 	return success;
 }
 
+bool phongo_execute_bulkwritecommand(zval* manager, php_phongo_bulkwritecommand_t* bwc, zval* zoptions, uint32_t server_id, zval* return_value)
+{
+	mongoc_client_t*                     client = NULL;
+	mongoc_bulkwrite_t*                  bw = bwc->bw;
+	mongoc_bulkwriteopts_t*              bw_opts = NULL;
+	mongoc_bulkwritereturn_t             bw_ret = { 0 };
+	php_phongo_bulkwritecommandresult_t* bwcr;
+	zval*                                zsession = NULL;
+	bool                                 success = true;
+
+	client = Z_MANAGER_OBJ_P(manager)->client;
+
+	if (!phongo_parse_session(zoptions, client, NULL, &zsession)) {
+		/* Exception should already have been thrown */
+		return false;
+	}
+
+	mongoc_bulkwrite_set_client(bw, client);
+
+	bw_opts = phongo_bwc_assemble_opts(bwc);
+	mongoc_bulkwriteopts_set_serverid(bw_opts, server_id);
+
+	if (zsession) {
+		mongoc_bulkwrite_set_session(bw, Z_SESSION_OBJ_P(zsession)->client_session);
+		/* Save a reference to the session on the class struct to avoid leaving
+		 * a dangling pointer within mongoc_bulkwrite_t. */
+		ZVAL_ZVAL(&bwc->session, zsession, 1, 0);
+	}
+
+	bw_ret = mongoc_bulkwrite_execute(bw, bw_opts);
+
+	bwcr = phongo_bulkwritecommandresult_init(return_value, &bw_ret, manager);
+
+	if (bw_ret.exc) {
+		success = false;
+		bson_error_t error = { 0 };
+
+		// Check if there is a top-level error
+		if (mongoc_bulkwriteexception_error(bw_ret.exc, &error)) {
+			phongo_throw_exception_from_bson_error_t_and_reply(&error, mongoc_bulkwriteexception_errorreply(bw_ret.exc));
+		} else {
+			// TODO: Determine appropriate message w/o top-level error
+			zend_throw_exception(php_phongo_bulkwritecommandexception_ce, "Bulk write failed", 0);
+		}
+
+		/* Unlike mongoc_bulk_operation_execute, mongoc_bulkwrite_execute may
+		 * report COMMAND_INVALID_ARG alongside a partial result (CDRIVER-5842).
+		 * If there is no result, we can throw InvalidArgumentException without
+		 * proxying it behind a BulkWriteException. */
+		if (!bw_ret.res && error.domain == MONGOC_ERROR_COMMAND && error.code == MONGOC_ERROR_COMMAND_INVALID_ARG) {
+			// TODO: Do we care about other mongoc_bulkwriteexception_t fields?
+			goto cleanup;
+		}
+
+		if (EG(exception)) {
+			char* message;
+
+			(void) spprintf(&message, 0, "Bulk write failed due to previous %s: %s", PHONGO_ZVAL_EXCEPTION_NAME(EG(exception)), error.message);
+			zend_throw_exception(php_phongo_bulkwritecommandexception_ce, message, 0);
+			efree(message);
+		}
+
+		/* Ensure error labels are added to the final BulkWriteCommandException. If a
+		 * previous exception was also thrown, error labels will already have
+		 * been added by phongo_throw_exception_from_bson_error_t_and_reply. */
+		phongo_exception_add_error_labels(mongoc_bulkwriteexception_errorreply(bw_ret.exc));
+		phongo_add_exception_prop(ZEND_STRL("bulkWriteCommandResult"), return_value);
+	}
+
+cleanup:
+	mongoc_bulkwriteopts_destroy(bw_opts);
+	mongoc_bulkwriteresult_destroy(bw_ret.res);
+	mongoc_bulkwriteexception_destroy(bw_ret.exc);
+
+	return success;
+}
+
 bool phongo_execute_command(zval* manager, php_phongo_command_type_t type, const char* db, zval* zcommand, zval* options, uint32_t server_id, zval* return_value)
 {
 	mongoc_client_t*            client;

src/MongoDB/Server.stub.php

@@ -22,6 +22,8 @@
 
 #include <php.h>
 
+#include "phongo_structs.h"
+
 /* This enum is used for processing options and selecting a libmongoc function
  * to use in phongo_execute_command. The values are important, as READ and WRITE
  * are also used as a bit field to determine whether readPreference,
@@ -37,6 +39,7 @@ typedef enum {
 } php_phongo_command_type_t;
 
 bool phongo_execute_bulk_write(zval* manager, const char* namespace, php_phongo_bulkwrite_t* bulk_write, zval* zwriteConcern, uint32_t server_id, zval* return_value);
+bool phongo_execute_bulkwritecommand(zval* manager, php_phongo_bulkwritecommand_t* bwc, zval* zoptions, uint32_t server_id, zval* return_value);
 bool phongo_execute_command(zval* manager, php_phongo_command_type_t type, const char* db, zval* zcommand, zval* zreadPreference, uint32_t server_id, zval* return_value);
 bool phongo_execute_query(zval* manager, const char* namespace, zval* zquery, zval* zreadPreference, uint32_t server_id, zval* return_value);