Add ability to pass cache_statement per operation (#352)

Previously ecto_sql used pre-defined cache_statement value for all
types of queries. This commit adds ability to use user specified
cache statement as it could improve performance in certain situations
like multiple `insert_all` queries with fixed size batch and so on.

Besides this the commit adds length of parameters to the default
cache_statement name. In other way set of queries like this:

    App.Repo.insert(%MyModel{name: "test_1", field_id: nil})
    App.Repo.insert(%MyModel{name: "test_2", field_id: "test"})
    App.Repo.insert(%MyModel{name: "test_3", field_id: nil})

will lead to situation when the previous prepared statement will be
closed when next query from the set will be executed and new one will
be created because `nil` values are not get into resulted query.

Author: 0xAX

lib/ecto/adapters/myxql.ex

@@ -256,7 +256,11 @@ defmodule Ecto.Adapters.MyXQL do
     key = primary_key!(schema_meta, returning)
     {fields, values} = :lists.unzip(params)
     sql = @conn.insert(prefix, source, fields, [fields], on_conflict, [], [])
-    opts = [{:cache_statement, "ecto_insert_#{source}"} | opts]
+    opts = if is_nil(Keyword.get(opts, :cache_statement)) do
+      [{:cache_statement, "ecto_insert_#{source}_#{length(fields)}"} | opts]
+    else
+      opts
+    end
 
     case Ecto.Adapters.SQL.query(adapter_meta, sql, values ++ query_params, opts) do
       {:ok, %{num_rows: 1, last_insert_id: last_insert_id}} ->

lib/ecto/adapters/sql.ex

@@ -656,7 +656,11 @@ defmodule Ecto.Adapters.SQL do
 
     sql = conn.insert(prefix, source, header, rows, on_conflict, returning, placeholders)
 
-    opts = [{:cache_statement, "ecto_insert_all_#{source}"} | opts]
+    opts = if is_nil(Keyword.get(opts, :cache_statement)) do
+      [{:cache_statement, "ecto_insert_all_#{source}"} | opts]
+    else
+      opts
+    end
 
     all_params = placeholders ++ Enum.reverse(params, conflict_params)
     %{num_rows: num, rows: rows} =
@@ -789,9 +793,13 @@ defmodule Ecto.Adapters.SQL do
 
   @doc false
   def struct(adapter_meta, conn, sql, operation, source, params, values, on_conflict, returning, opts) do
-    cache_statement = "ecto_#{operation}_#{source}"
+    opts = if is_nil(Keyword.get(opts, :cache_statement)) do
+      [{:cache_statement, "ecto_#{operation}_#{source}_#{length(params)}"} | opts]
+    else
+      opts
+    end
 
-    case query(adapter_meta, sql, values, [cache_statement: cache_statement] ++ opts) do
+    case query(adapter_meta, sql, values, opts) do
       {:ok, %{rows: nil, num_rows: 1}} ->
         {:ok, []}