Merge pull request #65 from hanumantmk/writeErrorFix

Bulk API fixes

Author: chergert

src/mongoc/mongoc-write-command-private.h

@@ -54,6 +54,7 @@ typedef struct
          bson_t   *documents;
          uint32_t  n_documents;
          uint32_t  n_merged;
+         uint32_t  current_n_documents;
       } insert;
       struct {
          uint8_t   ordered : 1;

src/mongoc/mongoc-write-command.c

@@ -53,7 +53,10 @@ typedef void (*mongoc_write_op_t) (mongoc_write_command_t       *command,
 
 static bson_t gEmptyWriteConcern = BSON_INITIALIZER;
 
-
+static int32_t
+_mongoc_write_result_merge_arrays (mongoc_write_result_t *result,
+                                   bson_t                *dest,
+                                   bson_iter_t           *iter);
 void
 _mongoc_write_command_insert_append (mongoc_write_command_t *command,
                                      const bson_t * const   *documents,
@@ -379,6 +382,7 @@ _mongoc_write_command_insert_legacy (mongoc_write_command_t       *command,
 
 cleanup:
    if (gle) {
+      command->u.insert.current_n_documents = i;
       _mongoc_write_result_merge_legacy (result, command, gle);
       bson_destroy (gle);
       gle = NULL;
@@ -651,6 +655,7 @@ _mongoc_write_command_insert (mongoc_write_command_t       *command,
       result->failed = true;
    }
 
+   command->u.insert.current_n_documents = i;
    _mongoc_write_result_merge (result, command, &reply);
 
    bson_destroy (&cmd);
@@ -876,6 +881,7 @@ _mongoc_write_result_merge_legacy (mongoc_write_result_t  *result, /* IN */
                                    const bson_t           *reply)  /* IN */
 {
    const bson_value_t *value;
+   bson_t holder, write_errors, child;
    bson_iter_t iter;
    bson_iter_t ar;
    bson_iter_t citer;
@@ -909,6 +915,21 @@ _mongoc_write_result_merge_legacy (mongoc_write_result_t  *result, /* IN */
                       code,
                       "%s", err);
       result->failed = true;
+
+      bson_init(&holder);
+      bson_append_array_begin(&holder, "0", 1, &write_errors);
+      bson_append_document_begin(&write_errors, "0", 1, &child);
+      bson_append_int32(&child, "index", 5, 0);
+      bson_append_int32(&child, "code", 4, code);
+      bson_append_utf8(&child, "errmsg", 6, err, -1);
+      bson_append_document_end(&write_errors, &child);
+      bson_append_array_end(&holder, &write_errors);
+      bson_iter_init(&iter, &holder);
+      bson_iter_next(&iter);
+
+      _mongoc_write_result_merge_arrays (result, &result->writeErrors, &iter);
+
+      bson_destroy(&holder);
    }
 
    switch (command->type) {
@@ -974,7 +995,7 @@ _mongoc_write_result_merge_legacy (mongoc_write_result_t  *result, /* IN */
       result->offset += 1;
       break;
    case MONGOC_WRITE_COMMAND_INSERT:
-      result->offset += command->u.insert.n_documents;
+      result->offset += command->u.insert.current_n_documents;
       break;
    default:
       break;
@@ -1144,7 +1165,7 @@ _mongoc_write_result_merge (mongoc_write_result_t  *result,  /* IN */
       result->offset += affected;
       break;
    case MONGOC_WRITE_COMMAND_INSERT:
-      result->offset += command->u.insert.n_documents;
+      result->offset += command->u.insert.current_n_documents;
       break;
    default:
       break;

tests/test-bulk.c

@@ -291,6 +291,74 @@ test_index_offset (void)
    bson_destroy (sel);
 }
 
+static void
+test_bulk_edge_over_1000 (void)
+{
+   mongoc_client_t *client;
+   mongoc_collection_t *collection;
+   mongoc_bulk_operation_t * bulk_op;
+   mongoc_write_concern_t * wc = mongoc_write_concern_new();
+   bson_iter_t iter, error_iter, index;
+   bson_t doc, result;
+   bson_error_t error;
+   int i;
+
+   client = mongoc_client_new (gTestUri);
+   assert (client);
+
+   collection = get_test_collection (client, "OVER_1000");
+   assert (collection);
+
+   mongoc_write_concern_set_w(wc, 1);
+
+   bulk_op = mongoc_collection_create_bulk_operation(collection, false, wc);
+
+   for (i = 0; i < 1010; i+=3) {
+      bson_init(&doc);
+      bson_append_int32(&doc, "_id", -1, i);
+
+      mongoc_bulk_operation_insert(bulk_op, &doc);
+
+      bson_destroy(&doc);
+   }
+
+   mongoc_bulk_operation_execute(bulk_op, NULL, &error);
+
+   mongoc_bulk_operation_destroy(bulk_op);
+
+   bulk_op = mongoc_collection_create_bulk_operation(collection, false, wc);
+   for (i = 0; i < 1010; i++) {
+      bson_init(&doc);
+      bson_append_int32(&doc, "_id", -1, i);
+
+      mongoc_bulk_operation_insert(bulk_op, &doc);
+
+      bson_destroy(&doc);
+   }
+
+   mongoc_bulk_operation_execute(bulk_op, &result, &error);
+
+   bson_iter_init_find(&iter, &result, "writeErrors");
+   assert(bson_iter_recurse(&iter, &error_iter));
+   assert(bson_iter_next(&error_iter));
+
+   for (i = 0; i < 1010; i+=3) {
+      assert(bson_iter_recurse(&error_iter, &index));
+      assert(bson_iter_find(&index, "index"));
+      if (bson_iter_int32(&index) != i) {
+          fprintf(stderr, "index should be %d, but is %d\n", i, bson_iter_int32(&index));
+      }
+      assert(bson_iter_int32(&index) == i);
+      bson_iter_next(&error_iter);
+   }
+
+   mongoc_bulk_operation_destroy(bulk_op);
+
+   mongoc_write_concern_destroy(wc);
+
+   mongoc_collection_destroy(collection);
+   mongoc_client_destroy(client);
+}
 
 static void
 test_bulk_edge_case_372 (void)
@@ -497,6 +565,7 @@ test_bulk_install (TestSuite *suite)
    TestSuite_Add (suite, "/BulkOperation/index_offset", test_index_offset);
    TestSuite_Add (suite, "/BulkOperation/CDRIVER-372", test_bulk_edge_case_372);
    TestSuite_Add (suite, "/BulkOperation/new", test_bulk_new);
+   TestSuite_Add (suite, "/BulkOperation/over_1000", test_bulk_edge_over_1000);
 
    atexit (cleanup_globals);
 }