BulkWriteResult defstruct has a new attribute errors, which contains the errors reported from the server.

Author: zookzook

lib/mongo/bulk_write.ex

@@ -200,33 +200,10 @@ defmodule Mongo.BulkWrite do
       ops
       |> get_op_sequence()
       |> Enum.map(fn {cmd, docs} -> one_bulk_write_operation(conn, cmd, coll, docs, write_concern, max_batch_size, opts) end)
-      |> Enum.reduce(empty, fn
-        {cmd, {count, ids}}, acc         -> merge(cmd, count, ids, acc)
-        {cmd, {mat, mod, ups, ids}}, acc -> merge(cmd, mat, mod, ups, ids, acc)
-        {cmd, count}, acc                -> merge(cmd, count, acc)
-
-      end)
+      |> BulkWriteResult.reduce(empty)
     end
   end
 
-  defp merge(:insert, count, ids, %BulkWriteResult{:inserted_count => value, :inserted_ids => current_ids} = result) do
-    %BulkWriteResult{result | inserted_count: value + count, inserted_ids: current_ids ++ ids}
-  end
-  defp merge(:update, matched, modified, upserts, ids,
-                      %BulkWriteResult{:matched_count => matched_count,
-                       :modified_count => modified_count,
-                       :upserted_count => upserted_count,
-                       :upserted_ids => current_ids} = result) do
-    %BulkWriteResult{result | matched_count: matched_count + matched,
-               modified_count: modified_count + modified,
-               upserted_count: upserted_count + upserts,
-               upserted_ids: current_ids ++ ids}
-  end
-  defp merge(:delete, count, %BulkWriteResult{:deleted_count => value} = result) do
-    %BulkWriteResult{result | deleted_count: value + count}
-  end
-  defp merge(_other, _count, result), do: result
-
   ##
   # returns the current write concerns from `opts`
   #
@@ -252,20 +229,12 @@ defmodule Mongo.BulkWrite do
 
     with {:ok, limits} <- Mongo.limits(conn),
          max_batch_size <- limits.max_write_batch_size,
-         {_, {inserts, ids}}                           <- one_bulk_write_operation(conn, :insert, coll, inserts, write_concern, max_batch_size, opts),
-         {_, {matched, modified, upserts, upsert_ids}} <- one_bulk_write_operation(conn, :update, coll, updates, write_concern, max_batch_size, opts),
-         {_, deletes}                                  <- one_bulk_write_operation(conn, :delete, coll, deletes, write_concern, max_batch_size, opts) do
-
-      %BulkWriteResult{
-        acknowledged: acknowledged(write_concern),
-        inserted_count: inserts,
-        inserted_ids: ids,
-        matched_count: matched,
-        deleted_count: deletes,
-        modified_count: modified,
-        upserted_count: upserts,
-        upserted_ids: upsert_ids
-      }
+         insert_result <- one_bulk_write_operation(conn, :insert, coll, inserts, write_concern, max_batch_size, opts),
+         update_result <- one_bulk_write_operation(conn, :update, coll, updates, write_concern, max_batch_size, opts),
+         delete_result <- one_bulk_write_operation(conn, :delete, coll, deletes, write_concern, max_batch_size, opts) do
+
+      [insert_result, update_result, delete_result]
+      |> BulkWriteResult.reduce(%BulkWriteResult{acknowledged: acknowledged(write_concern)})
     end
   end
 
@@ -274,7 +243,7 @@ defmodule Mongo.BulkWrite do
   #
   defp one_bulk_write_operation(conn, cmd, coll, docs, write_concern, max_batch_size, opts) do
     with result <- conn |> run_commands(get_cmds(cmd, coll, docs, write_concern, max_batch_size, opts), opts) |> collect(cmd) do
-      {cmd, result}
+      result
     end
   end
 
@@ -330,35 +299,34 @@ defmodule Mongo.BulkWrite do
   #
   defp collect({docs, ids}, :insert) do
 
-    {docs
-     |> Enum.map(fn
-       {:ok, %{"n" => n}} -> n
-       {:ok, _other}      -> 0
-     end)
-     |> Enum.reduce(0, fn x, acc -> x + acc end), ids}
-
+    docs
+    |> Enum.map(fn
+      {:ok, %{"n" => n} = doc} -> BulkWriteResult.insert_result(n, ids, doc["writeErrors"] || [])
+      {:ok, _other}            -> BulkWriteResult.empty()
+    end)
+    |> BulkWriteResult.reduce()
   end
+
   defp collect(docs, :update) do
 
     docs
     |> Enum.map(fn
-      {:ok, %{"n" => n, "nModified" => modified, "upserted" => ids}} -> l = length(ids); {n - l, modified, l, filter_upsert_ids(ids)}
-      {:ok, %{"n" => matched, "nModified" => modified}}              -> {matched, modified, 0, []}
-      {:ok, _other}                                                  -> {0, 0, 0, []}
-    end)
-    |> Enum.reduce({0, 0, 0, []}, fn
-      {mat, mod, ups, ids}, {s_mat, s_mod, s_ups, all} -> {s_mat + mat, s_mod + mod, s_ups + ups, all ++ ids}
+      {:ok, %{"n" => n, "nModified" => modified, "upserted" => ids} = doc} -> l = length(ids)
+                                                                              BulkWriteResult.update_result(n - l, modified, l, filter_upsert_ids(ids), doc["writeErrors"] || [])
+      {:ok, %{"n" => matched, "nModified" => modified} = doc}              -> BulkWriteResult.update_result(matched, modified, 0, [], doc["writeErrors"] || [])
+      {:ok, _other}                                                        -> BulkWriteResult.empty()
     end)
+    |> BulkWriteResult.reduce()
 
   end
   defp collect(docs, :delete) do
 
     docs
     |> Enum.map(fn
-      {:ok, %{"n" => n}} -> n
-      {:ok, _other}      -> 0
+      {:ok, %{"n" => n} = doc } -> BulkWriteResult.delete_result(n, doc["writeErrors"] || [])
+      {:ok, _other}             -> BulkWriteResult.empty()
     end)
-    |> Enum.reduce(0, fn x, acc -> x + acc end)
+    |> BulkWriteResult.reduce()
 
   end
 

lib/mongo/results.ex

@@ -76,6 +76,8 @@ defmodule Mongo.BulkWriteResult do
     * `:upserted_count` - Number of upserted documents
     * `:upserted_ids` - If the operation was an upsert, the upserted ids
     * `:inserted_ids` - If the operation was an insert, the inserted ids
+    * `:errors` - If the operation results in an error, the error is collected
+
   """
 
   @type t :: %__MODULE__{
@@ -86,9 +88,49 @@ defmodule Mongo.BulkWriteResult do
                deleted_count: non_neg_integer,
                upserted_count: non_neg_integer,
                upserted_ids: list(BSON.ObjectId.t),
-               inserted_ids: list(BSON.ObjectId.t)
+               inserted_ids: list(BSON.ObjectId.t),
+               errors: list(Map.t)
              }
 
-  defstruct [acknowledged: true, matched_count: 0, modified_count: 0, inserted_count: 0, deleted_count: 0, upserted_count: 0, inserted_ids: [], upserted_ids: []]
+   alias Mongo.BulkWriteResult
+
+  defstruct [acknowledged: true, matched_count: 0, modified_count: 0, inserted_count: 0, deleted_count: 0, upserted_count: 0, inserted_ids: [], upserted_ids: [], errors: []]
+
+  def insert_result(count, ids, errors) do
+    %BulkWriteResult{inserted_count: count, inserted_ids: ids, errors: errors}
+  end
+
+  def update_result(matched_count, modified_count, upserted_count, ids, errors) do
+    %BulkWriteResult{matched_count: matched_count, modified_count: modified_count, upserted_count: upserted_count, upserted_ids: ids, errors: errors}
+  end
+
+  def delete_result(count, errors) do
+    %BulkWriteResult{deleted_count: count, errors: errors}
+  end
+
+  def empty() do
+    %BulkWriteResult{}
+  end
+
+  def add(%BulkWriteResult{} = src, %BulkWriteResult{} = dest) do
+    %BulkWriteResult{acknowledged: src.acknowledged,
+      matched_count: src.matched_count + dest.matched_count,
+      modified_count: src.modified_count + dest.modified_count,
+      inserted_count: src.inserted_count + dest.inserted_count,
+      deleted_count: src.deleted_count + dest.deleted_count,
+      upserted_count: src.upserted_count + dest.upserted_count,
+      inserted_ids: src.inserted_ids ++ dest.inserted_ids,
+      upserted_ids: src.upserted_ids ++ dest.upserted_ids,
+      errors: src.errors ++ dest.errors
+    }
+  end
+
+  def reduce(results, acc) do
+    Enum.reduce(results, acc, fn x, acc -> BulkWriteResult.add(acc, x) end)
+  end
+
+  def reduce(results) do
+    reduce(results, %BulkWriteResult{})
+  end
 
 end

test/mongo/bulk_writes_test.exs

@@ -143,5 +143,33 @@ defmodule Mongo.BulkWritesTest do
 
   end
 
+  test "create one small document and one too large document", top do
+    coll = unique_name()
+    max_n = (16*1024*1024)
+
+    a_line_1k = Enum.reduce(1..1_024, "", fn _, acc -> acc <> "A" end)
+    a_line_1m = Enum.reduce(1..1_024, "", fn _, acc -> acc <> a_line_1k end)
+    a_line_16m = String.slice(Enum.reduce(1..16, "", fn _, acc -> acc <> a_line_1m end), 0..max_n)
+
+    b_line_1k = Enum.reduce(1..1_024, "", fn _, acc -> acc <> "B" end)
+    b_line_1m = Enum.reduce(1..1_024, "", fn _, acc -> acc <> b_line_1k end)
+    b_line_16m = String.slice(Enum.reduce(1..15, "", fn _, acc -> acc <> b_line_1m end), 0..max_n)
+
+    bulk = coll
+           |> OrderedBulk.new()
+           |> OrderedBulk.insert_one(%{v: a_line_1k, key: "small"})
+           |> OrderedBulk.insert_one(%{v: a_line_16m, key: "big"})
+           |> OrderedBulk.update_one(%{key: "small"}, %{"$set": %{v: b_line_1k}})
+           |> OrderedBulk.update_one(%{key: "big"}, %{"$set": %{v: b_line_16m}})
+           |> OrderedBulk.delete_one(%{key: "small"})
+           |> OrderedBulk.delete_one(%{key: "big"})
+
+    %BulkWriteResult{errors: [%{"code" => code}]} = result = BulkWrite.write(top.pid, bulk, w: 1)
+
+    assert code == 2
+    assert %{:matched_count => 1, :deleted_count => 1, :modified_count => 1} ==  Map.take(result, [:matched_count, :deleted_count, :modified_count])
+    assert {:ok, 0} == Mongo.count(top.pid, coll, %{})
+
+  end
 
 end