Add adapter level read-only transaction support

lib/ecto/adapters/myxql/connection.ex

@@ -27,6 +27,12 @@ if Code.ensure_loaded?(MyXQL) do
       MyXQL.query(conn, sql, params, opts)
     end
 
+    @impl true
+    def read_only_transaction(conn, opts, fun) do
+      {:ok, _} = query(conn, "SET TRANSACTION READ ONLY", [], log: opts[:log])
+      DBConnection.transaction(conn, fun, opts)
+    end
+
     @impl true
     def query_many(conn, sql, params, opts) do
       ensure_list_params!(params)

lib/ecto/adapters/postgres/connection.ex

@@ -131,6 +131,18 @@ if Code.ensure_loaded?(Postgrex) do
       Postgrex.query(conn, sql, params, opts)
     end
 
+    @impl true
+    def read_only_transaction(conn, opts, fun) do
+      DBConnection.transaction(
+        conn,
+        fn transaction_conn ->
+          {:ok, _} = query(transaction_conn, "SET TRANSACTION READ ONLY", [], log: opts[:log])
+          fun.(transaction_conn)
+        end,
+        opts
+      )
+    end
+
     @impl true
     def query_many(_conn, _sql, _params, _opts) do
       raise RuntimeError, "query_many is not supported in the PostgreSQL adapter"

lib/ecto/adapters/sql.ex

@@ -278,7 +278,7 @@ defmodule Ecto.Adapters.SQL do
 
       @impl true
       def transaction(meta, opts, fun) do
-        Ecto.Adapters.SQL.transaction(meta, opts, fun)
+        Ecto.Adapters.SQL.transaction(meta, @conn, opts, fun)
       end
 
       @impl true
@@ -1214,8 +1214,12 @@ defmodule Ecto.Adapters.SQL do
   ## Transactions
 
   @doc false
-  def transaction(adapter_meta, opts, callback) do
-    checkout_or_transaction(:transaction, adapter_meta, opts, callback)
+  def transaction(adapter_meta, connection, opts, callback) do
+    if opts[:read_only] do
+      read_only_transaction(adapter_meta, connection, opts, callback)
+    else
+      checkout_or_transaction(:transaction, adapter_meta, opts, callback)
+    end
   end
 
   @doc false
@@ -1468,6 +1472,33 @@ defmodule Ecto.Adapters.SQL do
 
   ## Connection helpers
 
+  defp read_only_transaction(adapter_meta, connection, opts, callback) do
+    %{pid: pool, telemetry: telemetry, opts: default_opts, log_stacktrace_mfa: log_stacktrace_mfa} =
+      adapter_meta
+
+    opts = with_log(telemetry, log_stacktrace_mfa, [], opts ++ default_opts)
+
+    callback = fn transaction_conn ->
+      previous_conn = put_conn(pool, transaction_conn)
+
+      try do
+        callback.()
+      after
+        reset_conn(pool, previous_conn)
+      end
+    end
+
+    if function_exported?(connection, :read_only_transaction, 3) do
+      DBConnection.run(
+        get_conn_or_pool(pool, adapter_meta),
+        &connection.read_only_transaction(&1, opts, callback),
+        opts
+      )
+    else
+      raise ArgumentError, "read-only transactions are not supported by #{inspect(connection)}"
+    end
+  end
+
   defp checkout_or_transaction(fun, adapter_meta, opts, callback) do
     %{pid: pool, telemetry: telemetry, opts: default_opts, log_stacktrace_mfa: log_stacktrace_mfa} =
       adapter_meta

lib/ecto/adapters/sql/connection.ex

@@ -45,6 +45,15 @@ defmodule Ecto.Adapters.SQL.Connection do
   @callback query_many(connection, statement, params, options :: Keyword.t()) ::
               {:ok, term} | {:error, Exception.t()}
 
+  @doc """
+  Runs the given function inside a read-only transaction.
+  """
+  @callback read_only_transaction(connection, options :: Keyword.t(), (connection -> result)) ::
+              {:ok, result} | {:error, term()}
+            when result: var
+
+  @optional_callbacks read_only_transaction: 3
+
   @doc """
   Returns a stream that prepares and executes the given query with
   `DBConnection`.

lib/mix/tasks/ecto.query.ex

@@ -81,12 +81,15 @@ defmodule Mix.Tasks.Ecto.Query do
       if opts[:sql] do
         {:ok, format_sql(repo, query)}
       else
-        read_only_transaction(repo, fn ->
-          query
-          |> repo.all()
-          |> Enum.take(limit)
-          |> inspect_entries()
-        end)
+        repo.transaction(
+          fn ->
+            query
+            |> repo.all()
+            |> Enum.take(limit)
+            |> inspect_entries()
+          end,
+          read_only: true
+        )
       end
 
     result
@@ -145,35 +148,6 @@ defmodule Mix.Tasks.Ecto.Query do
   defp alias?({:alias, _, [_aliases, _opts]}), do: true
   defp alias?(_), do: false
 
-  defp read_only_transaction(repo, fun) do
-    do_read_only_transaction(repo.__adapter__(), repo, fun)
-  end
-
-  defp do_read_only_transaction(Ecto.Adapters.Postgres, repo, fun) do
-    repo.transaction(fn ->
-      repo.query!("SET TRANSACTION READ ONLY", [], log: false)
-      fun.()
-    end)
-  end
-
-  defp do_read_only_transaction(Ecto.Adapters.MyXQL, repo, fun) do
-    repo.checkout(fn ->
-      repo.query!("START TRANSACTION READ ONLY", [], log: false)
-
-      try do
-        {:ok, fun.()}
-      after
-        repo.query!("ROLLBACK", [], log: false)
-      end
-    end)
-  end
-
-  defp do_read_only_transaction(adapter, _repo, _fun) do
-    Mix.raise(
-      "ecto.query requires read-only transactions, which are not supported by #{inspect(adapter)}"
-    )
-  end
-
   defp format_sql(repo, query) do
     {sql, params} = repo.to_sql(:all, query)
 

test/mix/tasks/ecto.query_test.exs

@@ -52,9 +52,8 @@ defmodule Mix.Tasks.Ecto.QueryTest do
 
       run(["-r", to_string(__MODULE__.PostgresRepo), "from(p in Post)"])
 
-      assert_received {:transaction, __MODULE__.PostgresRepo, _fun}
-      assert_received {:query!, "SET TRANSACTION READ ONLY", [], opts}
-      assert opts[:log] == false
+      assert_received {:transaction, __MODULE__.PostgresRepo, _fun, opts}
+      assert opts[:read_only]
       assert_received {:all, %Ecto.Query{}}
       assert_received {:mix_shell, :info, [output]}
 
@@ -78,7 +77,8 @@ defmodule Mix.Tasks.Ecto.QueryTest do
 
       run(["from(p in Post)"])
 
-      assert_received {:query!, "SET TRANSACTION READ ONLY", [], _opts}
+      assert_received {:transaction, __MODULE__.PostgresRepo, _fun, opts}
+      assert opts[:read_only]
     end)
   end
 
@@ -196,25 +196,22 @@ defmodule Mix.Tasks.Ecto.QueryTest do
 
       assert_received {:to_sql, :all, %Ecto.Query{}, []}
       refute_received {:all, _query}
-      refute_received {:query!, "SET TRANSACTION READ ONLY", [], _opts}
+      refute_received {:transaction, __MODULE__.PostgresRepo, _fun, _opts}
       assert_received {:mix_shell, :info, [output]}
 
       assert output =~ ~s[SQL:\nSELECT p0."id" FROM "posts" AS p0 WHERE (p0."id" = $1)]
       assert output =~ "Params:\n[1]"
     end)
   end
 
-  test "starts MySQL read-only transactions explicitly" do
+  test "runs MySQL queries in a read-only transaction" do
     Process.put(:test_repo_all_results, [[1, "first", "hunter2"]])
 
     run(["-r", to_string(__MODULE__.MyXQLRepo), inspect(Post)])
 
-    assert_received {:checkout, __MODULE__.MyXQLRepo}
-    assert_received {:myxql_query!, "START TRANSACTION READ ONLY", [], opts}
-    assert opts[:log] == false
+    assert_received {:myxql_transaction, __MODULE__.MyXQLRepo, _fun, opts}
+    assert opts[:read_only]
     assert_received {:myxql_all, %Ecto.Query{}}
-    assert_received {:myxql_query!, "ROLLBACK", [], rollback_opts}
-    assert rollback_opts[:log] == false
     assert_received {:mix_shell, :info, [output]}
     assert output =~ ~s(title: "first")
   end
@@ -224,9 +221,7 @@ defmodule Mix.Tasks.Ecto.QueryTest do
 
     run(["-r", to_string(__MODULE__.MyXQLRepo), "--sql", inspect(Post)])
 
-    refute_received {:checkout, __MODULE__.MyXQLRepo}
-    refute_received {:myxql_query!, "START TRANSACTION READ ONLY", [], _opts}
-    refute_received {:myxql_query!, "ROLLBACK", [], _opts}
+    refute_received {:myxql_transaction, __MODULE__.MyXQLRepo, _fun, _opts}
     assert_received {:myxql_to_sql, :all, %Ecto.Query{}, []}
     assert_received {:mix_shell, :info, [output]}
     assert output =~ ~s[SQL:\nSELECT p0.`id` FROM `posts` AS p0]
@@ -278,20 +273,24 @@ defmodule Mix.Tasks.Ecto.QueryTest do
   end
 
   test "raises when the adapter does not support read-only transactions" do
-    assert_raise Mix.Error, ~r/read-only transactions.*Ecto.Adapters.Tds/s, fn ->
+    assert_raise ArgumentError, ~r/read-only transactions are not supported/, fn ->
       run(["-r", to_string(__MODULE__.NoReadOnlyRepo), inspect(Post)])
     end
   end
 
   defmodule NoReadOnlyRepo do
     def __adapter__, do: Ecto.Adapters.Tds
+
+    def transaction(_fun, _opts) do
+      raise ArgumentError, "read-only transactions are not supported"
+    end
   end
 
   defmodule PostgresRepo do
     def __adapter__, do: Ecto.Adapters.Postgres
 
-    def transaction(fun, _opts \\ []) do
-      send(self(), {:transaction, __MODULE__, fun})
+    def transaction(fun, opts \\ []) do
+      send(self(), {:transaction, __MODULE__, fun, opts})
       {:ok, fun.()}
     end
 
@@ -326,14 +325,9 @@ defmodule Mix.Tasks.Ecto.QueryTest do
   defmodule MyXQLRepo do
     def __adapter__, do: Ecto.Adapters.MyXQL
 
-    def checkout(fun, _opts \\ []) do
-      send(test_process(), {:checkout, __MODULE__})
-      fun.()
-    end
-
-    def query!(sql, params \\ [], opts \\ []) do
-      send(test_process(), {:myxql_query!, sql, params, opts})
-      %{rows: [], num_rows: 0}
+    def transaction(fun, opts \\ []) do
+      send(test_process(), {:myxql_transaction, __MODULE__, fun, opts})
+      {:ok, fun.()}
     end
 
     def all(query) do