Ecto.Adapters.SQL.stream/2 does not consider dynamic repo

[frerich]

Elixir version

1.19.4

Database and Version

PostgreSQL 17.7

Ecto Versions

ecto 3.13.5, ecto_sql 3.13.2

Database Adapter and Versions (postgrex, myxql, etc)

postgrex 0.21.1

Current behavior

This small script attempts to stream a single line of text into a database table while using a dynamic repo:

Mix.install([
  {:ecto_sql, "~> 3.10"},
  {:postgrex, ">= 0.0.0"}
])

Application.put_env(:myapp, Repo, database: "example_db")

defmodule Repo do
  use Ecto.Repo, adapter: Ecto.Adapters.Postgres, otp_app: :myapp
end

# Bring up the Repo
Repo.__adapter__().storage_up(Repo.config())
{:ok, _pid} = Repo.start_link(Repo.config())

# Now, the actual test case.
default_dyn_repo  = Repo.get_dynamic_repo()

{:ok, dyn_repo} = Repo.config() |> Keyword.put(:name, nil) |> Repo.start_link()

Repo.put_dynamic_repo(dyn_repo)

Repo.transaction(fn ->
  ["data"]
  |> Stream.into(Ecto.Adapters.SQL.stream(Repo, "COPY tmp_table FROM STDIN"))
  |> Stream.run()
end)

Repo.put_dynamic_repo(default_dyn_repo)

It currently fails with

** (RuntimeError) cannot collect into stream outside of transaction
    (ecto_sql 3.13.3) lib/ecto/adapters/sql.ex:1152: Ecto.Adapters.SQL.into/4
    (ecto_sql 3.13.3) lib/ecto/adapters/sql/stream.ex:29: Collectable.Ecto.Adapters.SQL.Stream.into/1
    (elixir 1.19.4) lib/stream.ex:589: Stream.do_into/5
    (elixir 1.19.4) lib/stream.ex:707: Stream.run/1
    repo.exs:9: anonymous fn/1 in Repo.transaction/2
    (ecto 3.13.5) lib/ecto/repo/transaction.ex:11: anonymous fn/3 in Ecto.Repo.Transaction.transact/4
    (ecto_sql 3.13.3) lib/ecto/adapters/sql.ex:1468: anonymous fn/3 in Ecto.Adapters.SQL.checkout_or_transaction/4
    (db_connection 2.8.1) lib/db_connection.ex:1753: DBConnection.run_transaction/4

To fix this, I currently have to pass dyn_repo (the PID of the dynamic repo) resp. Repo.get_dynamic_repo() to Ecto.Adapters.SQL.stream/2 explicitly, instead of the module Repo. It is unclear if the desired (if so: undocumented) requirement, or if it's a missing feature in Ecto.Adapters.SQL.stream/2 to call get_dynamic_repo/0 itself.

Expected behavior

The Ecto.Adapters.SQL.stream/2 call should respect the dynamic repo. I.e. the example program should notice that there is an active transaction and start streaming, which should then run into the error

** (Postgrex.Error) ERROR 42P01 (undefined_table) relation "tmp_table" does not exist
    (db_connection 2.8.1) lib/db_connection.ex:848: DBConnection.execute!/4
    (postgrex 0.21.1) lib/postgrex/stream.ex:85: anonymous fn/7 in Collectable.Postgrex.Stream.make_into/4
    (ecto_sql 3.13.3) lib/ecto/adapters/sql/stream.ex:36: anonymous fn/4 in Collectable.Ecto.Adapters.SQL.Stream.make_into/2
    (elixir 1.19.4) lib/stream.ex:612: Stream.do_into/4
    (elixir 1.19.4) lib/stream.ex:707: Stream.run/1
    repo.exs:9: anonymous fn/1 in Repo.transaction/2
    (ecto 3.13.5) lib/ecto/repo/transaction.ex:11: anonymous fn/3 in Ecto.Repo.Transaction.transact/4
    (ecto_sql 3.13.3) lib/ecto/adapters/sql.ex:1468: anonymous fn/3 in Ecto.Adapters.SQL.checkout_or_transaction/4

[greg-rychlewski]

I think it only works right now if you do this Repo.stream("COPY tmp_table FROM STDIN") instead of this Ecto.Adapters.SQL.stream(Repo, "COPY tmp_table FROM STDIN")

Would you be able to give that a try?

[frerich]

Repo.stream/1 won't work, no. My understanding is that it can only be used for streaming out of a database - it expects that it's given a queryable yielding records.

Trying to use it in the example program raises:

** (Ecto.QueryError) queries that do not have a schema need to explicitly pass a :select clause in query:

from c0 in "COPY tmp_table FROM STDIN"

    (ecto 3.13.5) lib/ecto/query/planner.ex:2736: Ecto.Query.Planner.error!/2
    (ecto 3.13.5) lib/ecto/repo/queryable.ex:37: Ecto.Repo.Queryable.stream/3
    repo.exs:25: anonymous fn/0 in :elixir_compiler_0.__FILE__/1
    repo.exs:9: anonymous fn/1 in Repo.transaction/2
    (ecto 3.13.5) lib/ecto/repo/transaction.ex:11: anonymous fn/3 in Ecto.Repo.Transaction.transact/4
    (ecto_sql 3.13.3) lib/ecto/adapters/sql.ex:1468: anonymous fn/3 in Ecto.Adapters.SQL.checkout_or_transaction/4
    (db_connection 2.8.1) lib/db_connection.ex:1753: DBConnection.run_transaction/4
    repo.exs:23: (file)

For streaming into a database using e.g. COPY ... FROM STDIN, I believe Ecto.Adapters.SQL.stream/2 has to be used.

[greg-rychlewski]

Oh I see, yes you're right. At the moment none of the Ecto.Adapters.SQL.X functions that take repo as a first argument get the dynamic repo. They all take it at face value. Only the Repo.X functions do so.

So I believe you have to pass get_dynamic_repo() as the first argument. And then the question is do we improve the documentation to make this clearer or is this something that should get the dynamic repo. For that I need @josevalim to weigh in. I'm guessing it's a documentation fix but could be wrong.

[josevalim]

Ecto.Adapters.SQL.stream is a callback which is not meant to be invoked directly. You can see it is not even documented: https://hexdocs.pm/ecto_sql/Ecto.Adapters.SQL.html

So yes, you should use Repo.stream instead!

[frerich]

I'd happily use Repo.stream instead - if I knew how to use it for streamng data into the database via e.g. COPY ... FROM STDIN as I'm doing with Ecto.Adapters.SQL.stream/2 above. Any idea?

[josevalim]

Sorry, I clearly lied. stream/4 is public, so we have to fix this.

[josevalim]

Unfortunately we cannot make it just work, because the Ecto.Adapters.SQL functions receive processes, so we cannot assume an atom represents a repo, it represents a process name.

The docs currently say:

You can also invoke them directly from Ecto.Adapters.SQL, but
keep in mind that in such cases features such as "dynamic repositories"
won't be available

Does calling this work for you:

Ecto.Adapters.SQL.stream(Repo.get_dynamic_repo(), "COPY tmp_table FROM STDIN")

?

[frerich]

Ah - the atom does not represent a repo, it's a process name! 💡

Yes, passing Repo.get_dynamic_repo() works in my case, but I understand that's because it either returns an atom which matches the name of the Repo process in case no dynamic repo is set, or it returns a PID (in case a dynamic repo is set). Both cases work.

Passing Repo.get_dynamic_repo() works for me, maybe this is just a piece of missing documentation then.