Workaround and tests for Bulk API edge case with pre-2.6 servers

Author: samantharitter

lib/mongo/bulk_write_collection_view.rb

@@ -297,6 +297,13 @@ def merge_result(errors, exchanges)
             upserted = [{"_id" => upserted}] if upserted.class != Array # OP_UPDATE non-array
             n_upserted = upserted.size
             concat(result, "upserted", merge_indexes(upserted, exchange))
+          elsif (response["updatedExisting"] == false && n == 1)
+            # workaround for DRIVERS-151 (non-ObjectID _id fields in pre-2.6 servers)
+            op = exchange[:batch][0]
+            missing_id = op[:u].fetch(:_id, op[:q][:_id]) # _id in update document takes precedence
+            upserted = [ { "_id" => missing_id, "index" => 0 } ]
+            n_upserted = n
+            concat(result, "upserted", merge_indexes(upserted, exchange))
           end
           tally(result, "nUpserted", n_upserted) if n_upserted > 0
           tally(result, "nMatched", n - n_upserted)

test/functional/bulk_write_collection_view_test.rb

@@ -501,6 +501,27 @@ def nil_tally_responses(responses, key)
       end
     end
 
+    should "count nUpserted correctly when _id is not an ObjectId (upsert-update)" do
+      with_write_commands_and_operations(@db.connection) do |wire_version|
+        @collection.remove
+
+        bulk = @collection.initialize_unordered_bulk_op
+        bulk.find({:_id => 3}).upsert.update({"$set" => {:b => 3}})
+        result = bulk.execute
+        assert_match_document(
+            {
+                "ok" => 1,
+                "n" => 1,
+                "nMatched" => 0,
+                "nUpserted" => 1,
+                "nModified" => batch_commands?(wire_version) ? 0 : nil,
+                "upserted" => [
+                    { "_id" => 3, "index" => 0 }
+                ]
+            }, result, "wire_version:#{wire_version}")
+      end
+    end
+
     # ----- UPSERT-UPDATE_ONE -----
 
     should "#upsert a document without affecting non-upsert update_ones" do
@@ -544,6 +565,27 @@ def nil_tally_responses(responses, key)
       end
     end
 
+
+    should "count nUpserted correctly when _id is not an ObjectId (upsert-update_one)" do
+      with_write_commands_and_operations(@db.connection) do |wire_version|
+        @collection.remove
+        bulk = @collection.initialize_ordered_bulk_op
+        bulk.find({:_id => 2}).upsert.update_one({"$set" => {:x => 2}})
+        result = bulk.execute
+        assert_match_document(
+            {
+                "ok" => 1,
+                "n" => 1,
+                "nMatched" => 0,
+                "nUpserted" => 1,
+                "nModified" => batch_commands?(wire_version) ? 0 : nil,
+                "upserted" => [
+                    {"_id" => 2, "index" => 0 }
+                ]
+            }, result, "wire_version:#{wire_version}")
+      end
+    end
+
     # ----- UPSERT-REPLACE_ONE -----
 
     should "not affect non-upsert replace_ones in same batch as #upsert-replace_one" do
@@ -598,6 +640,28 @@ def nil_tally_responses(responses, key)
       assert_equal({"nM" => nil}, nil_tally_responses([{"nM" => 1}, {"nM" => nil}, {"nM" => 3}], "nM"))
     end
 
+
+    should "count nUpserted correctly when _id is not an ObjectId (upsert-replace_one)" do
+      with_write_commands_and_operations(@db.connection) do |wire_version|
+        @collection.remove
+        bulk = @collection.initialize_unordered_bulk_op
+        bulk.find({:a => 1}).upsert.replace_one({:_id => 2})
+        result = bulk.execute
+        assert_match_document(
+            {
+                "ok" => 1,
+                "n" => 1,
+                "nMatched" => 0,
+                "nUpserted" => 1,
+                "nModified" => batch_commands?(wire_version) ? 0 : nil,
+                "upserted" => [
+                                { "_id" => 2, "index" => 0 }
+                ]
+            }, result, "wire_version:#{wire_version}")
+        assert_equal 1, @collection.count
+      end
+    end
+
     # ----- MIXED OPS, ORDERED -----
 
     should "execute, return result and reset @ops for #execute" do