text: /*
   * Copyright 2014 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.
   */
  
  
  #include "mongoc/mongoc-bulk-operation.h"
  #include "mongoc/mongoc-bulk-operation-private.h"
  #include "mongoc/mongoc-client-private.h"
  #include "mongoc/mongoc-trace-private.h"
  #include "mongoc/mongoc-write-concern-private.h"
  #include "mongoc/mongoc-util-private.h"
  #include "mongoc/mongoc-opts-private.h"
  #include "mongoc/mongoc-write-command-private.h"
  
  
  /*
   * This is the implementation of both write commands and bulk write commands.
   * They are all implemented as one contiguous set since we'd like to cut down
   * on code duplication here.
   *
   * This implementation is currently naive.
   *
   * Some interesting optimizations might be:
   *
   *   - If unordered mode, send operations as we get them instead of waiting
   *     for execute() to be called. This could save us memcpy()'s too.
   *   - If there is no acknowledgement desired, keep a count of how many
   *     replies we need and ask the socket layer to skip that many bytes
   *     when reading.
   *   - Try to use iovec to send write commands with subdocuments rather than
   *     copying them into the write command document.
   */
  
  
  mongoc_bulk_operation_t *
  mongoc_bulk_operation_new (bool ordered)
  {
     mongoc_bulk_operation_t *bulk;
  
     bulk = (mongoc_bulk_operation_t *) bson_malloc0 (sizeof *bulk);
     bulk->flags.bypass_document_validation = false;
     bulk->flags.ordered = ordered;
     bulk->server_id = 0;
  
     _mongoc_array_init (&bulk->commands, sizeof (mongoc_write_command_t));
     _mongoc_write_result_init (&bulk->result);
  
     return bulk;
  }
  
  
  mongoc_bulk_operation_t *
  _mongoc_bulk_operation_new (
     mongoc_client_t *client,                     /* IN */
     const char *database,                        /* IN */
     const char *collection,                      /* IN */
     mongoc_bulk_write_flags_t flags,             /* IN */
     const mongoc_write_concern_t *write_concern) /* IN */
  {
     mongoc_bulk_operation_t *bulk;
  
     BSON_ASSERT (client);
     BSON_ASSERT (collection);
  
     bulk = mongoc_bulk_operation_new (flags.ordered);
     bulk->client = client;
     bulk->database = bson_strdup (database);
     bulk->collection = bson_strdup (collection);
     bulk->write_concern = mongoc_write_concern_copy (write_concern);
     bulk->executed = false;
     bulk->flags = flags;
     bulk->operation_id = ++client->cluster.operation_id;
  
     return bulk;
  }
  
  
  void
  mongoc_bulk_operation_destroy (mongoc_bulk_operation_t *bulk) /* IN */
  {
     mongoc_write_command_t *command;
     int i;
  
     if (bulk) {
        for (i = 0; i < bulk->commands.len; i++) {
           command =
              &_mongoc_array_index (&bulk->commands, mongoc_write_command_t, i);
           _mongoc_write_command_destroy (command);
        }
  
        bson_free (bulk->database);
        bson_free (bulk->collection);
        mongoc_write_concern_destroy (bulk->write_concern);
        _mongoc_array_destroy (&bulk->commands);
  
        _mongoc_write_result_destroy (&bulk->result);
  
        bson_free (bulk);
     }
  }
  
  
  /* already failed, e.g. a bad call to mongoc_bulk_operation_insert? */
  #define BULK_EXIT_IF_PRIOR_ERROR       \
     do {                                \
        if (bulk->result.error.domain) { \
           EXIT;                         \
        }                                \
     } while (0)
  
  #define BULK_RETURN_IF_PRIOR_ERROR                                            \
     do {                                                                       \
        if (bulk->result.error.domain) {                                        \
           if (error != &bulk->result.error) {                                  \
              bson_set_error (error,                                            \
                              MONGOC_ERROR_COMMAND,                             \
                              MONGOC_ERROR_COMMAND_INVALID_ARG,                 \
                              "Bulk operation is invalid from prior error: %s", \
                              bulk->result.error.message);                      \
           };                                                                   \
           return false;                                                        \
        };                                                                      \
     } while (0)
  
  
  bool
  _mongoc_bulk_operation_remove_with_opts (
     mongoc_bulk_operation_t *bulk,
     const bson_t *selector,
     const mongoc_bulk_remove_opts_t *remove_opts,
     int32_t limit,
     bson_error_t *error) /* OUT */
  {
     mongoc_write_command_t command = {0};
     mongoc_write_command_t *last;
     bson_t opts;
     bool has_collation;
     bool ret = false;
  
     ENTRY;
  
     BSON_ASSERT (bulk);
     BSON_ASSERT (selector);
  
     bson_init (&opts);
  
     /* allow "limit" in opts, but it must be the correct limit */
     if (remove_opts->limit != limit) {
        bson_set_error (error,
                        MONGOC_ERROR_COMMAND,
                        MONGOC_ERROR_COMMAND_INVALID_ARG,
                        "Invalid \"limit\" in opts: %" PRId32 "."
                        " The value must be %" PRId32 ", or omitted.",
                        remove_opts->limit,
                        limit);
        GOTO (done);
     }
  
     bson_append_int32 (&opts, "limit", 5, limit);
     has_collation = !bson_empty (&remove_opts->collation);
     if (has_collation) {
        bson_append_document (&opts, "collation", 9, &remove_opts->collation);
     }
  
     if (bulk->commands.len) {
        last = &_mongoc_array_index (
           &bulk->commands, mongoc_write_command_t, bulk->commands.len - 1);
        if (last->type == MONGOC_WRITE_COMMAND_DELETE) {
           last->flags.has_collation |= has_collation;
           last->flags.has_multi_write |= (remove_opts->limit == 0);
           _mongoc_write_command_delete_append (last, selector, &opts);
           ret = true;
           GOTO (done);
        }
     }
  
     _mongoc_write_command_init_delete (
        &command, selector, NULL, &opts, bulk->flags, bulk->operation_id);
  
     command.flags.has_collation = has_collation;
     command.flags.has_multi_write = (remove_opts->limit == 0);
  
     _mongoc_array_append_val (&bulk->commands, command);
     ret = true;
  
  done:
     bson_destroy (&opts);
     RETURN (ret);
  }
  
  
  bool
  mongoc_bulk_operation_remove_one_with_opts (mongoc_bulk_operation_t *bulk,
                                              const bson_t *selector,
                                              const bson_t *opts,
                                              bson_error_t *error) /* OUT */
  {
     mongoc_bulk_remove_one_opts_t remove_opts;
     bool ret;
  
     ENTRY;
  
     BULK_RETURN_IF_PRIOR_ERROR;
  
     if (!_mongoc_bulk_remove_one_opts_parse (
            bulk->client, opts, &remove_opts, error)) {
        _mongoc_bulk_remove_one_opts_cleanup (&remove_opts);
        RETURN (false);
     }
  
     ret = _mongoc_bulk_operation_remove_with_opts (
        bulk, selector, &remove_opts.remove, 1, error);
  
     _mongoc_bulk_remove_one_opts_cleanup (&remove_opts);
     RETURN (ret);
  }
  
  
  bool
  mongoc_bulk_operation_remove_many_with_opts (mongoc_bulk_operation_t *bulk,
                                               const bson_t *selector,
                                               const bson_t *opts,
                                               bson_error_t *error) /* OUT */
  {
     mongoc_bulk_remove_many_opts_t remove_opts;
     bool ret;
  
     ENTRY;
  
     BULK_RETURN_IF_PRIOR_ERROR;
  
     if (!_mongoc_bulk_remove_many_opts_parse (
            bulk->client, opts, &remove_opts, error)) {
        _mongoc_bulk_remove_many_opts_cleanup (&remove_opts);
        RETURN (false);
     }
  
     ret = _mongoc_bulk_operation_remove_with_opts (
        bulk, selector, &remove_opts.remove, 0, error);
  
     _mongoc_bulk_remove_many_opts_cleanup (&remove_opts);
     RETURN (ret);
  }
  
  
  void
  mongoc_bulk_operation_remove (mongoc_bulk_operation_t *bulk, /* IN */
                                const bson_t *selector)        /* IN */
  {
     bson_error_t *error = &bulk->result.error;
  
     ENTRY;
  
     BULK_EXIT_IF_PRIOR_ERROR;
  
     if (!mongoc_bulk_operation_remove_many_with_opts (
            bulk, selector, NULL, error)) {
        MONGOC_WARNING ("%s", error->message);
     }
  
     if (error->domain) {
        MONGOC_WARNING ("%s", error->message);
     }
  
     EXIT;
  }
  
  
  void
  mongoc_bulk_operation_remove_one (mongoc_bulk_operation_t *bulk, /* IN */
                                    const bson_t *selector)        /* IN */
  {
     bson_error_t *error = &bulk->result.error;
  
     ENTRY;
  
     BULK_EXIT_IF_PRIOR_ERROR;
  
     if (!mongoc_bulk_operation_remove_one_with_opts (
            bulk, selector, NULL, error)) {
        MONGOC_WARNING ("%s", error->message);
     }
  
     if (error->domain) {
        MONGOC_WARNING ("%s", error->message);
     }
  
     EXIT;
  }
  
  void
  mongoc_bulk_operation_delete (mongoc_bulk_operation_t *bulk,
                                const bson_t *selector)
  {
     ENTRY;
  
     mongoc_bulk_operation_remove (bulk, selector);
  
     EXIT;
  }
  
  void
  mongoc_bulk_operation_delete_one (mongoc_bulk_operation_t *bulk,
                                    const bson_t *selector)
  {
     ENTRY;
  
     mongoc_bulk_operation_remove_one (bulk, selector);
  
     EXIT;
  }
  
  void
  mongoc_bulk_operation_insert (mongoc_bulk_operation_t *bulk,
                                const bson_t *document)
  {
     ENTRY;
  
     BSON_ASSERT (bulk);
     BSON_ASSERT (document);
  
     if (!mongoc_bulk_operation_insert_with_opts (
            bulk, document, NULL /* opts */, &bulk->result.error)) {
        MONGOC_WARNING ("%s", bulk->result.error.message);
     }
  
     EXIT;
  }
  
  bool
  mongoc_bulk_operation_insert_with_opts (mongoc_bulk_operation_t *bulk,
                                          const bson_t *document,
                                          const bson_t *opts,
                                          bson_error_t *error)
  {
     mongoc_bulk_insert_opts_t insert_opts;
     mongoc_write_command_t command = {0};
     mongoc_write_command_t *last;
     bool ret = false;
  
     ENTRY;
  
     BSON_ASSERT (bulk);
     BSON_ASSERT (document);
  
     BULK_RETURN_IF_PRIOR_ERROR;
  
     if (!_mongoc_bulk_insert_opts_parse (
            bulk->client, opts, &insert_opts, error)) {
        GOTO (done);
     }
  
     if (!_mongoc_validate_new_document (document, insert_opts.validate, error)) {
        GOTO (done);
     }
  
     if (bulk->commands.len) {
        last = &_mongoc_array_index (
           &bulk->commands, mongoc_write_command_t, bulk->commands.len - 1);
  
        if (last->type == MONGOC_WRITE_COMMAND_INSERT) {
           _mongoc_write_command_insert_append (last, document);
           ret = true;
           GOTO (done);
        }
     }
  
     _mongoc_write_command_init_insert (
        &command,
        document,
        opts,
        bulk->flags,
        bulk->operation_id,
        !mongoc_write_concern_is_acknowledged (bulk->write_concern));
  
     _mongoc_array_append_val (&bulk->commands, command);
  
     ret = true;
  
  done:
     _mongoc_bulk_insert_opts_cleanup (&insert_opts);
     RETURN (ret);
  }
  
  static void
  _mongoc_bulk_operation_update_append (
     mongoc_bulk_operation_t *bulk,
     const bson_t *selector,
     const bson_t *document,
     const mongoc_bulk_update_opts_t *update_opts,
     const bson_t *array_filters,
     const bson_t *extra_opts)
  {
     mongoc_write_command_t command = {0};
     mongoc_write_command_t *last;
     bson_t opts;
     bool has_collation;
     bool has_array_filters;
  
     bson_init (&opts);
     bson_append_bool (&opts, "upsert", 6, update_opts->upsert);
     bson_append_bool (&opts, "multi", 5, update_opts->multi);
  
     has_array_filters = !bson_empty0 (array_filters);
     if (has_array_filters) {
        bson_append_document (&opts, "arrayFilters", 12, array_filters);
     }
  
     has_collation = !bson_empty (&update_opts->collation);
     if (has_collation) {
        bson_append_document (&opts, "collation", 9, &update_opts->collation);
     }
  
     if (extra_opts) {
        bson_concat (&opts, extra_opts);
     }
  
     if (bulk->commands.len) {
        last = &_mongoc_array_index (
           &bulk->commands, mongoc_write_command_t, bulk->commands.len - 1);
        if (last->type == MONGOC_WRITE_COMMAND_UPDATE) {
           last->flags.has_collation |= has_collation;
           last->flags.has_multi_write |= update_opts->multi;
           _mongoc_write_command_update_append (last, selector, document, &opts);
           bson_destroy (&opts);
           return;
        }
     }
  
     _mongoc_write_command_init_update (
        &command, selector, document, &opts, bulk->flags, bulk->operation_id);
  
     command.flags.has_array_filters = has_array_filters;
     command.flags.has_collation = has_collation;
     command.flags.has_multi_write = update_opts->multi;
  
     _mongoc_array_append_val (&bulk->commands, command);
     bson_destroy (&opts);
  }
  
  static bool
  _mongoc_bulk_operation_update_with_opts (
     mongoc_bulk_operation_t *bulk,
     const bson_t *selector,
     const bson_t *document,
     const mongoc_bulk_update_opts_t *update_opts,
     const bson_t *array_filters,
     const bson_t *extra_opts,
     bool multi,
     bson_error_t *error) /* OUT */
  {
     ENTRY;
  
     BSON_ASSERT (bulk);
     BSON_ASSERT (selector);
     BSON_ASSERT (document);
  
     if (!_mongoc_validate_update (document, update_opts->validate, error)) {
        RETURN (false);
     }
  
     /* allow "multi" in opts, but it must be the correct multi */
     if (update_opts->multi != multi) {
        bson_set_error (error,
                        MONGOC_ERROR_COMMAND,
                        MONGOC_ERROR_COMMAND_INVALID_ARG,
                        "Invalid \"multi\" in opts: %s."
                        " The value must be %s, or omitted.",
                        update_opts->multi ? "true" : "false",
                        multi ? "true" : "false");
        RETURN (false);
     }
  
     _mongoc_bulk_operation_update_append (
        bulk, selector, document, update_opts, array_filters, extra_opts);
  
     RETURN (true);
  }
  
  bool
  mongoc_bulk_operation_update_one_with_opts (mongoc_bulk_operation_t *bulk,
                                              const bson_t *selector,
                                              const bson_t *document,
                                              const bson_t *opts,
                                              bson_error_t *error) /* OUT */
  {
     mongoc_bulk_update_one_opts_t update_opts;
     bool ret;
  
     ENTRY;
  
     BULK_RETURN_IF_PRIOR_ERROR;
  
     if (!_mongoc_bulk_update_one_opts_parse (
            bulk->client, opts, &update_opts, error)) {
        _mongoc_bulk_update_one_opts_cleanup (&update_opts);
        RETURN (false);
     }
  
     ret = _mongoc_bulk_operation_update_with_opts (bulk,
                                                    selector,
                                                    document,
                                                    &update_opts.update,
                                                    &update_opts.arrayFilters,
                                                    &update_opts.extra,
                                                    false /* multi */,
                                                    error);
  
     _mongoc_bulk_update_one_opts_cleanup (&update_opts);
     RETURN (ret);
  }
  
  bool
  mongoc_bulk_operation_update_many_with_opts (mongoc_bulk_operation_t *bulk,
                                               const bson_t *selector,
                                               const bson_t *document,
                                               const bson_t *opts,
                                               bson_error_t *error) /* OUT */
  {
     mongoc_bulk_update_many_opts_t update_opts;
     bool ret;
  
     ENTRY;
  
     BULK_RETURN_IF_PRIOR_ERROR;
  
     if (!_mongoc_bulk_update_many_opts_parse (
            bulk->client, opts, &update_opts, error)) {
        _mongoc_bulk_update_many_opts_cleanup (&update_opts);
        RETURN (false);
     }
  
     ret = _mongoc_bulk_operation_update_with_opts (bulk,
                                                    selector,
                                                    document,
                                                    &update_opts.update,
                                                    &update_opts.arrayFilters,
                                                    &update_opts.extra,
                                                    true /* multi */,
                                                    error);
  
     _mongoc_bulk_update_many_opts_cleanup (&update_opts);
     RETURN (ret);
  }
  
  void
  mongoc_bulk_operation_update (mongoc_bulk_operation_t *bulk,
                                const bson_t *selector,
                                const bson_t *document,
                                bool upsert)
  {
     bson_t opts;
     bson_error_t *error = &bulk->result.error;
  
     ENTRY;
  
     BULK_EXIT_IF_PRIOR_ERROR;
  
     bson_init (&opts);
     if (upsert) {
        BSON_APPEND_BOOL (&opts, "upsert", upsert);
     }
  
     if (!mongoc_bulk_operation_update_many_with_opts (
           bulk, selector, document, &opts, error)) {
        MONGOC_WARNING ("%s", error->message);
     }
  
     bson_destroy (&opts);
  
     if (error->domain) {
        MONGOC_WARNING ("%s", error->message);
     }
  
     EXIT;
  }
  
  void
  mongoc_bulk_operation_update_one (mongoc_bulk_operation_t *bulk,
                                    const bson_t *selector,
                                    const bson_t *document,
                                    bool upsert)
  {
     bson_t opts;
     bson_error_t *error = &bulk->result.error;
  
     ENTRY;
  
     BULK_EXIT_IF_PRIOR_ERROR;
  
     bson_init (&opts);
     BSON_APPEND_BOOL (&opts, "upsert", upsert);
  
     if (!mongoc_bulk_operation_update_one_with_opts (
            bulk, selector, document, &opts, error)) {
        MONGOC_WARNING ("%s", error->message);
     }
  
     bson_destroy (&opts);
  
     if (error->domain) {
        MONGOC_WARNING ("%s", error->message);
     }
  
     EXIT;
  }
  
  bool
  mongoc_bulk_operation_replace_one_with_opts (mongoc_bulk_operation_t *bulk,
                                               const bson_t *selector,
                                               const bson_t *document,
                                               const bson_t *opts,
                                               bson_error_t *error) /* OUT */
  {
     mongoc_bulk_replace_one_opts_t repl_opts;
     mongoc_bulk_update_opts_t *update_opts = &repl_opts.update;
     bool ret = false;
  
     ENTRY;
  
     BSON_ASSERT (bulk);
     BSON_ASSERT (selector);
     BSON_ASSERT (document);
  
     BULK_RETURN_IF_PRIOR_ERROR;
  
     if (!_mongoc_bulk_replace_one_opts_parse (
            bulk->client, opts, &repl_opts, error)) {
        GOTO (done);
     }
  
     if (!_mongoc_validate_replace (document, update_opts->validate, error)) {
        GOTO (done);
     }
  
     /* allow "multi" in opts, but it must be the correct multi */
     if (update_opts->multi) {
        bson_set_error (error,
                        MONGOC_ERROR_COMMAND,
                        MONGOC_ERROR_COMMAND_INVALID_ARG,
                        "Invalid \"multi\": true in opts for"
                        " mongoc_bulk_operation_replace_one_with_opts."
                        " The value must be true, or omitted.");
        GOTO (done);
     }
  
     _mongoc_bulk_operation_update_append (
        bulk, selector, document, update_opts, NULL, &repl_opts.extra);
     ret = true;
  
  done:
     _mongoc_bulk_replace_one_opts_cleanup (&repl_opts);
     RETURN (ret);
  }
  
  void
  mongoc_bulk_operation_replace_one (mongoc_bulk_operation_t *bulk,
                                     const bson_t *selector,
                                     const bson_t *document,
                                     bool upsert)
  {
     bson_t opts = BSON_INITIALIZER;
     bson_error_t *error = &bulk->result.error;
  
     ENTRY;
  
     BSON_APPEND_BOOL (&opts, "upsert", upsert);
  
     if (!mongoc_bulk_operation_replace_one_with_opts (
            bulk, selector, document, &opts, error)) {
        MONGOC_WARNING ("%s", error->message);
     }
  
     bson_destroy (&opts);
  
     EXIT;
  }
  
  uint32_t
  mongoc_bulk_operation_execute (mongoc_bulk_operation_t *bulk, /* IN */
                                 bson_t *reply,                 /* OUT */
                                 bson_error_t *error)           /* OUT */
  {
     mongoc_cluster_t *cluster;
     mongoc_write_command_t *command;
     mongoc_server_stream_t *server_stream;
     bool ret;
     uint32_t offset = 0;
     int i;
  
     ENTRY;
  
     BSON_ASSERT (bulk);
  
     if (!bulk->client) {
        bson_set_error (error,
                        MONGOC_ERROR_COMMAND,
                        MONGOC_ERROR_COMMAND_INVALID_ARG,
                        "mongoc_bulk_operation_execute() requires a client "
                        "and one has not been set.");
        GOTO (err);
     }
     cluster = &bulk->client->cluster;
  
     if (bulk->executed) {
        _mongoc_write_result_destroy (&bulk->result);
        _mongoc_write_result_init (&bulk->result);
     }
  
     bulk->executed = true;
  
     if (!bulk->database) {
        bson_set_error (error,
                        MONGOC_ERROR_COMMAND,
                        MONGOC_ERROR_COMMAND_INVALID_ARG,
                        "mongoc_bulk_operation_execute() requires a database "
                        "and one has not been set.");
        GOTO (err);
     } else if (!bulk->collection) {
        bson_set_error (error,
                        MONGOC_ERROR_COMMAND,
                        MONGOC_ERROR_COMMAND_INVALID_ARG,
                        "mongoc_bulk_operation_execute() requires a collection "
                        "and one has not been set.");
        GOTO (err);
     }
  
     /* error stored by functions like mongoc_bulk_operation_insert that
      * can't report errors immediately */
     if (bulk->result.error.domain) {
        if (error) {
           memcpy (error, &bulk->result.error, sizeof (bson_error_t));
        }
  
        GOTO (err);
     }
  
     if (!bulk->commands.len) {
        bson_set_error (error,
                        MONGOC_ERROR_COMMAND,
                        MONGOC_ERROR_COMMAND_INVALID_ARG,
                        "Cannot do an empty bulk write");
        GOTO (err);
     }
  
     for (i = 0; i < bulk->commands.len; i++) {
        if (bulk->server_id) {
           server_stream =
              mongoc_cluster_stream_for_server (cluster,
                                                bulk->server_id,
                                                true /* reconnect_ok */,
                                                bulk->session,
                                                reply,
                                                error);
        } else {
           server_stream = mongoc_cluster_stream_for_writes (
              cluster, bulk->session, reply, error);
        }
  
        if (!server_stream) {
           /* stream_for_server and stream_for_writes initialize reply on error */
           RETURN (false);
        }
  
        command =
           &_mongoc_array_index (&bulk->commands, mongoc_write_command_t, i);
  
        _mongoc_write_command_execute (command,
                                       bulk->client,
                                       server_stream,
                                       bulk->database,
                                       bulk->collection,
                                       bulk->write_concern,
                                       offset,
                                       bulk->session,
                                       &bulk->result);
  
        bulk->server_id = server_stream->sd->id;
  
        if (bulk->result.failed &&
            (bulk->flags.ordered || bulk->result.must_stop)) {
           mongoc_server_stream_cleanup (server_stream);
           GOTO (cleanup);
        }
  
        offset += command->n_documents;
        mongoc_server_stream_cleanup (server_stream);
     }
  
  cleanup:
     _mongoc_bson_init_if_set (reply);
     ret = MONGOC_WRITE_RESULT_COMPLETE (&bulk->result,
                                         bulk->client->error_api_version,
                                         bulk->write_concern,
                                         MONGOC_ERROR_COMMAND /* err domain */,
                                         reply,
                                         error);
  
     RETURN (ret ? bulk->server_id : 0);
  
  err:
     _mongoc_bson_init_if_set (reply);
     RETURN (false);
  }
  
  void
  mongoc_bulk_operation_set_write_concern (
     mongoc_bulk_operation_t *bulk, const mongoc_write_concern_t *write_concern)
  {
     BSON_ASSERT (bulk);
  
     if (bulk->write_concern) {
        mongoc_write_concern_destroy (bulk->write_concern);
     }
  
     if (write_concern) {
        bulk->write_concern = mongoc_write_concern_copy (write_concern);
     } else {
        bulk->write_concern = mongoc_write_concern_new ();
     }
  }
  
  const mongoc_write_concern_t *
  mongoc_bulk_operation_get_write_concern (const mongoc_bulk_operation_t *bulk)
  {
     BSON_ASSERT (bulk);
  
     return bulk->write_concern;
  }
  
  
  void
  mongoc_bulk_operation_set_database (mongoc_bulk_operation_t *bulk,
                                      const char *database)
  {
     BSON_ASSERT (bulk);
  
     if (bulk->database) {
        bson_free (bulk->database);
     }
  
     bulk->database = bson_strdup (database);
  }
  
  
  void
  mongoc_bulk_operation_set_collection (mongoc_bulk_operation_t *bulk,
                                        const char *collection)
  {
     BSON_ASSERT (bulk);
  
     if (bulk->collection) {
        bson_free (bulk->collection);
     }
  
     bulk->collection = bson_strdup (collection);
  }
  
  
  void
  mongoc_bulk_operation_set_client (mongoc_bulk_operation_t *bulk, void *client)
  {
     BSON_ASSERT (bulk);
     BSON_ASSERT (client);
  
     if (bulk->session) {
        BSON_ASSERT (bulk->session->client == client);
     }
  
     bulk->client = (mongoc_client_t *) client;
  
     /* if you call set_client, bulk was likely made by mongoc_bulk_operation_new,
      * not mongoc_collection_create_bulk_operation_with_opts(), so operation_id
      * is 0. */
     if (!bulk->operation_id) {
        bulk->operation_id = ++bulk->client->cluster.operation_id;
     }
  }
  
  
  void
  mongoc_bulk_operation_set_client_session (
     mongoc_bulk_operation_t *bulk,
     struct _mongoc_client_session_t *client_session)
  {
     BSON_ASSERT (bulk);
     BSON_ASSERT (client_session);
  
     if (bulk->client) {
        BSON_ASSERT (bulk->client == client_session->client);
     }
  
     bulk->session = client_session;
  }
  
  
  uint32_t
  mongoc_bulk_operation_get_hint (const mongoc_bulk_operation_t *bulk)
  {
     BSON_ASSERT (bulk);
  
     return bulk->server_id;
  }
  
  
  void
  mongoc_bulk_operation_set_hint (mongoc_bulk_operation_t *bulk,
                                  uint32_t server_id)
  {
     BSON_ASSERT (bulk);
  
     bulk->server_id = server_id;
  }
  
  
  void
  mongoc_bulk_operation_set_bypass_document_validation (
     mongoc_bulk_operation_t *bulk, bool bypass)
  {
     BSON_ASSERT (bulk);
  
     bulk->flags.bypass_document_validation = bypass;
  }
  