added max_write_batch_size from database

renamed direct_command to exec_command

Author: zookzook

lib/mongo.ex

@@ -53,7 +53,6 @@ defmodule Mongo do
   alias Mongo.TopologyDescription
   alias Mongo.Topology
   alias Mongo.UrlParser
-  alias Mongo.UnorderedBulk
 
   @timeout 15000 # 5000
 
@@ -290,7 +289,7 @@ defmodule Mongo do
     opts = Keyword.drop(opts, ~w(bypass_document_validation max_time projection return_document sort upsert collation)a)
 
     with {:ok, conn, _, _} <- select_server(topology_pid, :write, opts),
-         {:ok, doc} <- direct_command(conn, cmd, opts) do
+         {:ok, doc} <- exec_command(conn, cmd, opts) do
         {:ok, doc["value"]}
     end
 
@@ -333,7 +332,7 @@ defmodule Mongo do
     opts = Keyword.drop(opts, ~w(bypass_document_validation max_time projection return_document sort upsert collation)a)
 
     with {:ok, conn, _, _} <- select_server(topology_pid, :write, opts),
-         {:ok, doc} <- direct_command(conn, cmd, opts), do: {:ok, doc["value"]}
+         {:ok, doc} <- exec_command(conn, cmd, opts), do: {:ok, doc["value"]}
   end
 
   defp should_return_new(:after), do: true
@@ -364,7 +363,7 @@ defmodule Mongo do
     opts = Keyword.drop(opts, ~w(max_time projection sort collation)a)
 
     with {:ok, conn, _, _} <- select_server(topology_pid, :write, opts),
-         {:ok, doc} <- direct_command(conn, cmd, opts), do: {:ok, doc["value"]}
+         {:ok, doc} <- exec_command(conn, cmd, opts), do: {:ok, doc["value"]}
   end
 
   @doc false
@@ -468,7 +467,7 @@ defmodule Mongo do
 
     with {:ok, conn, slave_ok, _} <- select_server(topology_pid, :read, opts),
          opts = Keyword.put(opts, :slave_ok, slave_ok),
-         {:ok, doc} <- direct_command(conn, cmd, opts),
+         {:ok, doc} <- exec_command(conn, cmd, opts),
          do: {:ok, doc["values"]}
   end
 
@@ -567,13 +566,13 @@ defmodule Mongo do
     rp_opts = [read_preference: Keyword.get(opts, :read_preference, rp)]
     with {:ok, conn, slave_ok, _} <- select_server(topology_pid, :read, rp_opts),
          opts = Keyword.put(opts, :slave_ok, slave_ok),
-         do: direct_command(conn, cmd, opts)
+         do: exec_command(conn, cmd, opts)
   end
 
   @doc false
   ## refactor: exec_command
-  @spec direct_command(pid, BSON.document, Keyword.t) :: {:ok, BSON.document | nil} | {:error, Mongo.Error.t}
-  def direct_command(conn, cmd, opts) do
+  @spec exec_command(pid, BSON.document, Keyword.t) :: {:ok, BSON.document | nil} | {:error, Mongo.Error.t}
+  def exec_command(conn, cmd, opts) do
     action = %Query{action: :command}
 
     with {:ok, _cmd, doc} <- DBConnection.execute(conn, action, [cmd], defaults(opts)),
@@ -588,13 +587,25 @@ defmodule Mongo do
   @doc """
   Returns the current wire version.
   """
-  def wire_version(conn, opts \\ []) do
+  @spec wire_version(pid) :: {:ok, integer} | {:error, Mongo.Error.t}
+  def wire_version(conn) do
     cmd = %Query{action: :wire_version}
-    with {:ok, _cmd, version} <- DBConnection.execute(conn, cmd, %{}, defaults(opts)) do
+    with {:ok, _cmd, version} <- DBConnection.execute(conn, cmd, %{}, defaults([])) do
       {:ok, version}
     end
   end
 
+  @doc """
+  Returns the limits of the database.
+  """
+  @spec limits(pid) :: {:ok, BSON.document} | {:error, Mongo.Error.t}
+  def limits(conn) do
+    cmd = %Query{action: :limits}
+    with {:ok, _cmd, limits} <- DBConnection.execute(conn, cmd, %{}, defaults([])) do
+      {:ok, limits}
+    end
+  end
+
   @doc """
   Similar to `command/3` but unwraps the result and raises on error.
   """
@@ -641,7 +652,7 @@ defmodule Mongo do
     ] |> filter_nils()
 
     with {:ok, conn, _, _} <- select_server(topology_pid, :write, opts),
-         {:ok, doc} <- direct_command(conn, cmd, opts) do
+         {:ok, doc} <- exec_command(conn, cmd, opts) do
       case doc do
         %{"writeErrors" => _} -> {:error, %Mongo.WriteError{n: doc["n"], ok: doc["ok"], write_errors: doc["writeErrors"]}}
         _ ->
@@ -695,7 +706,7 @@ defmodule Mongo do
     ] |> filter_nils()
 
     with {:ok, conn, _, _} <- select_server(topology_pid, :write, opts),
-         {:ok, doc} <- direct_command(conn, cmd, opts) do
+         {:ok, doc} <- exec_command(conn, cmd, opts) do
       case doc do
         %{"writeErrors" => _} ->  {:error, %Mongo.WriteError{n: doc["n"], ok: doc["ok"], write_errors: doc["writeErrors"]}}
         _ ->
@@ -766,7 +777,7 @@ defmodule Mongo do
             ] |> filter_nils()
 
     with {:ok, conn, _, _} <- select_server(topology_pid, :write, opts),
-         {:ok, doc} <- direct_command(conn, cmd, opts) do
+         {:ok, doc} <- exec_command(conn, cmd, opts) do
       case doc do
         %{"writeErrors" => _} -> {:error, %Mongo.WriteError{n: doc["n"], ok: doc["ok"], write_errors: doc["writeErrors"]}}
         %{ "ok" => _ok, "n" => n } ->
@@ -886,7 +897,7 @@ defmodule Mongo do
             ] |> filter_nils()
 
     with {:ok, conn, _, _} <- select_server(topology_pid, :write, opts),
-         {:ok, doc}        <- direct_command(conn, cmd, opts) do
+         {:ok, doc}        <- exec_command(conn, cmd, opts) do
 
       case doc do
 

lib/mongo/bulk_write.ex

@@ -158,9 +158,6 @@ defmodule Mongo.BulkWrite do
   alias Mongo.UnorderedBulk
   alias Mongo.OrderedBulk
 
-  @max_batch_size 100_000   ## todo    The maxWriteBatchSize limit of a database, which indicates the maximum number of write operations permitted in a write batch, raises from 1,000 to 100,000.
-
-
   @doc """
   Executes unordered and ordered bulk writes.
 
@@ -204,10 +201,13 @@ defmodule Mongo.BulkWrite do
       insertedIds: [],
     }
 
-    with {:ok, conn, _, _} <- Mongo.select_server(topology_pid, :write, opts) do
+    with {:ok, conn, _, _} <- Mongo.select_server(topology_pid, :write, opts),
+         {:ok, limits} <- Mongo.limits(conn),
+         max_batch_size <- limits.max_write_batch_size do
 
-      get_op_sequence(coll, ops)
-      |> Enum.map(fn {cmd, docs} -> one_bulk_write_operation(conn, cmd, coll, docs, write_concern, opts) end)
+      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)
@@ -258,9 +258,11 @@ defmodule Mongo.BulkWrite do
   #
   defp one_bulk_write(conn, %UnorderedBulk{coll: coll, inserts: inserts, updates: updates, deletes: deletes}, write_concern, opts) do
 
-    with {_, {inserts, ids}}                           <- one_bulk_write_operation(conn, :insert, coll, inserts, write_concern, opts),
-         {_, {matched, modified, upserts, upsert_ids}} <- one_bulk_write_operation(conn, :update, coll, updates, write_concern, opts),
-         {_, deletes}                                  <- one_bulk_write_operation(conn, :delete, coll, deletes, write_concern, opts) 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
 
       %{
         acknowledged: acknowledged(write_concern),
@@ -278,18 +280,18 @@ defmodule Mongo.BulkWrite do
   ###
   # Executes the command `cmd` and collects the result.
   #
-  defp one_bulk_write_operation(conn, cmd, coll, docs, write_concern, opts) do
-    with result <- conn |> run_commands(get_cmds(cmd, coll, docs, write_concern, opts), opts) |> collect(cmd) 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}
     end
   end
 
   ##
   # Converts the list of operations into insert/update/delete commands
   #
-  defp get_cmds(:insert, coll, docs, write_concern, opts), do: get_insert_cmds(coll, docs, write_concern, opts)
-  defp get_cmds(:update, coll, docs, write_concern, opts), do: get_update_cmds(coll, docs, write_concern, opts)
-  defp get_cmds(:delete, coll, docs, write_concern, opts), do: get_delete_cmds(coll, docs, write_concern, opts)
+  defp get_cmds(:insert, coll, docs, write_concern, max_batch_size, opts), do: get_insert_cmds(coll, docs, write_concern, max_batch_size, opts)
+  defp get_cmds(:update, coll, docs, write_concern, max_batch_size, opts), do: get_update_cmds(coll, docs, write_concern, max_batch_size, opts)
+  defp get_cmds(:delete, coll, docs, write_concern, max_batch_size, opts), do: get_delete_cmds(coll, docs, write_concern, max_batch_size, opts)
 
   defp acknowledged(%{w: w}) when w > 0, do: true
   defp acknowledged(%{}), do: false
@@ -299,15 +301,15 @@ defmodule Mongo.BulkWrite do
   #
   # [inserts, inserts, updates] -> [[inserts, inserts],[updates]]
   #
-  defp get_op_sequence(coll, ops) do
-    get_op_sequence(coll, ops, [])
+  defp get_op_sequence(ops) do
+    get_op_sequence(ops, [])
   end
-  defp get_op_sequence(_coll, [], acc), do: acc
-  defp get_op_sequence(coll, ops, acc) do
+  defp get_op_sequence([], acc), do: acc
+  defp get_op_sequence(ops, acc) do
 
     [{kind, _doc} | _rest] = ops
-    {docs, rest} = find_max_sequence(kind, ops)
-    get_op_sequence(coll, rest, [{kind, docs} | acc])
+    {docs, rest}           = find_max_sequence(kind, ops)
+    get_op_sequence(rest, [{kind, docs} | acc])
 
   end
 
@@ -371,20 +373,19 @@ defmodule Mongo.BulkWrite do
   defp filter_upsert_ids(nil), do: []
   defp filter_upsert_ids(upserted), do: Enum.map(upserted, fn doc -> doc["_id"] end)
 
-
   defp run_commands(conn, {cmds, ids}, opts) do
-    {Enum.map(cmds, fn cmd -> Mongo.direct_command(conn, cmd, opts) end), ids}
+    {Enum.map(cmds, fn cmd -> Mongo.exec_command(conn, cmd, opts) end), ids}
   end
   defp run_commands(conn, cmds, opts) do
-    Enum.map(cmds, fn cmd -> Mongo.direct_command(conn, cmd, opts) end)
+    Enum.map(cmds, fn cmd -> Mongo.exec_command(conn, cmd, opts) end)
   end
 
-  defp get_insert_cmds(coll, docs, write_concern, _opts) do
+  defp get_insert_cmds(coll, docs, write_concern, max_batch_size, _opts) do
 
     {ids, docs} = assign_ids(docs)
 
     cmds = docs
-           |> Enum.chunk_every(@max_batch_size)
+           |> Enum.chunk_every(max_batch_size)
            |> Enum.map(fn inserts -> get_insert_cmd(coll, inserts, write_concern) end)
     {cmds, ids}
 
@@ -398,10 +399,10 @@ defmodule Mongo.BulkWrite do
 
   end
 
-  defp get_delete_cmds(coll, docs, write_concern, opts) do
+  defp get_delete_cmds(coll, docs, write_concern, max_batch_size, opts) do
 
     docs
-    |> Enum.chunk_every(@max_batch_size)
+    |> Enum.chunk_every(max_batch_size)
     |> Enum.map(fn deletes -> get_delete_cmd(coll, deletes, write_concern, opts) end)
 
   end
@@ -423,10 +424,10 @@ defmodule Mongo.BulkWrite do
 
   end
 
-  defp get_update_cmds(coll, docs, write_concern, opts) do
+  defp get_update_cmds(coll, docs, write_concern, max_batch_size, opts) do
 
     docs
-    |> Enum.chunk_every(@max_batch_size)
+    |> Enum.chunk_every(max_batch_size)
     |> Enum.map(fn updates -> get_update_cmd(coll, updates, write_concern, opts) end)
 
   end

lib/mongo/cursor.ex

@@ -44,7 +44,7 @@ defmodule Mongo.Cursor do
                      "cursor" => %{
                      "id" => cursor_id,
                      "ns" => coll,
-                     "firstBatch" => docs}}} when ok == 1 <- Mongo.direct_command(conn, cmd, opts) do
+                     "firstBatch" => docs}}} when ok == 1 <- Mongo.exec_command(conn, cmd, opts) do
             state(conn: conn, cursor: cursor_id, coll: coll, docs: docs)
         end
       end
@@ -95,8 +95,8 @@ defmodule Mongo.Cursor do
              "cursor" => %{
                "id" => cursor_id,
                "ns" => coll,
-               "firstBatch" => docs} = response}} when ok == 1 <- Mongo.direct_command(conn, cmd, opts),
-           {:ok, wire_version} <- Mongo.wire_version(conn, opts) do
+               "firstBatch" => docs} = response}} when ok == 1 <- Mongo.exec_command(conn, cmd, opts),
+           {:ok, wire_version} <- Mongo.wire_version(conn) do
 
           [%{"$changeStream" => stream_opts} | _pipeline] = Keyword.get(cmd, :pipeline) # extract the change stream options
 
@@ -138,7 +138,7 @@ defmodule Mongo.Cursor do
         maxTimeMS: opts[:max_time]
       ] |> filter_nils()
 
-      with {:ok, %{"cursor" => %{ "id" => cursor_id, "nextBatch" => docs}, "ok" => ok}} when ok == 1 <- Mongo.direct_command(conn, cmd, opts) do
+      with {:ok, %{"cursor" => %{ "id" => cursor_id, "nextBatch" => docs}, "ok" => ok}} when ok == 1 <- Mongo.exec_command(conn, cmd, opts) do
         {:ok, %{cursor_id: cursor_id, docs: docs}}
       end
 
@@ -158,7 +158,7 @@ defmodule Mongo.Cursor do
       with {:ok, %{"operationTime" => op_time,
                    "cursor" => %{"id" => new_cursor_id,
                                  "nextBatch" => docs} = cursor,
-                                 "ok" => ok}} when ok == 1 <- Mongo.direct_command(conn, get_more, opts) do
+                                 "ok" => ok}} when ok == 1 <- Mongo.exec_command(conn, get_more, opts) do
 
         old_token = change_stream(change_stream, :resume_token)
         change_stream = update_change_stream(change_stream, cursor["postBatchResumeToken"], op_time, List.last(docs))
@@ -175,7 +175,7 @@ defmodule Mongo.Cursor do
         {:error, %Mongo.Error{code: code} = not_resumable} when code == 11601 or code == 136 or code == 237 -> {:error, not_resumable}
         {:error, _error} ->
 
-          with {:ok, wire_version} <- Mongo.wire_version(conn, opts) do
+          with {:ok, wire_version} <- Mongo.wire_version(conn) do
 
             [%{"$changeStream" => stream_opts} | pipeline] = Keyword.get(aggregate_cmd, :pipeline) # extract the change stream options
 
@@ -281,7 +281,7 @@ defmodule Mongo.Cursor do
       with {:ok, %{"cursorsAlive" => [],
                    "cursorsNotFound" => [],
                    "cursorsUnknown" => [],
-                   "ok" => ok}} when ok == 1 <- Mongo.direct_command(conn, cmd, opts) do
+                   "ok" => ok}} when ok == 1 <- Mongo.exec_command(conn, cmd, opts) do
         :ok
       end
     end

lib/mongo/mongo_db_connection.ex

@@ -239,6 +239,9 @@ defmodule Mongo.MongoDBConnection do
     end
   end
 
+  defp execute_action(:limits, _, _, state) do
+    {:ok, state.limits, state}
+  end
   defp execute_action(:wire_version, _, _, state) do
     {:ok, state.wire_version, state}
   end

lib/mongo/monitor.ex

@@ -104,7 +104,7 @@ defmodule Mongo.Monitor do
   defp call_is_master(conn_pid, opts) do
     start_time = System.monotonic_time
     result = try do
-      Mongo.direct_command(conn_pid, [isMaster: 1], opts)
+      Mongo.exec_command(conn_pid, [isMaster: 1], opts)
     rescue
       e -> {:error, e}
     end

lib/mongo_db_connection/utils.ex

@@ -23,7 +23,7 @@ defmodule Mongo.MongoDBConnection.Utils do
 
   @doc """
     Invoking a command using connection stored in state, that means within a DBConnection call. Therefore
-    we cannot call DBConnect.execute() to reuse the command function in Monto.direct_command()
+    we cannot call DBConnect.execute() to reuse the command function in Monto.exec_command()
 
     Using op_query structure to invoke the command
   """