Updates structure dump to support multiple prefixes on pg and myxql (#490)

Author: Amish Patel

integration_test/myxql/storage_test.exs

@@ -150,7 +150,53 @@ defmodule Ecto.Integration.StorageTest do
     :ok = Ecto.Migrator.up(PoolRepo, num, Migration, log: false)
     {:ok, path} = Ecto.Adapters.MyXQL.structure_dump(tmp_path(), TestRepo.config())
     contents = File.read!(path)
-    assert contents =~ "INSERT INTO `schema_migrations` (version) VALUES ("
+    assert contents =~ "INSERT INTO `ecto_test`.`schema_migrations` (version) VALUES (#{num})"
+  end
+
+  test "dumps structure and schema_migration records from multiple prefixes" do
+    # Create the test_schema schema
+    create_database()
+    prefix = params()[:database]
+
+    # Run migrations
+    version = @base_migration + System.unique_integer([:positive])
+    :ok = Ecto.Migrator.up(PoolRepo, version, Migration, log: false)
+    :ok = Ecto.Migrator.up(PoolRepo, version, Migration, log: false, prefix: prefix)
+
+    config = Keyword.put(TestRepo.config(), :dump_prefixes, ["ecto_test", prefix])
+    {:ok, path} = Ecto.Adapters.MyXQL.structure_dump(tmp_path(), config)
+    contents = File.read!(path)
+
+    assert contents =~ "Current Database: `#{prefix}`"
+    assert contents =~ "Current Database: `ecto_test`"
+    assert contents =~ "CREATE TABLE `schema_migrations`"
+    assert contents =~ ~s[INSERT INTO `#{prefix}`.`schema_migrations` (version) VALUES (#{version})]
+    assert contents =~ ~s[INSERT INTO `ecto_test`.`schema_migrations` (version) VALUES (#{version})]
+  after
+    drop_database()
+  end
+
+  test "dumps structure and schema_migration records only from queried prefix" do
+    # Create the test_schema schema
+    create_database()
+    prefix = params()[:database]
+
+    # Run migrations
+    version = @base_migration + System.unique_integer([:positive])
+    :ok = Ecto.Migrator.up(PoolRepo, version, Migration, log: false)
+    :ok = Ecto.Migrator.up(PoolRepo, version, Migration, log: false, prefix: prefix)
+
+    config = Keyword.put(TestRepo.config(), :dump_prefixes, ["ecto_test"])
+    {:ok, path} = Ecto.Adapters.MyXQL.structure_dump(tmp_path(), config)
+    contents = File.read!(path)
+
+    refute contents =~ "Current Database: `#{prefix}`"
+    assert contents =~ "Current Database: `ecto_test`"
+    assert contents =~ "CREATE TABLE `schema_migrations`"
+    refute contents =~ ~s[INSERT INTO `#{prefix}`.`schema_migrations` (version) VALUES (#{version})]
+    assert contents =~ ~s[INSERT INTO `ecto_test`.`schema_migrations` (version) VALUES (#{version})]
+  after
+    drop_database()
   end
 
   defp strip_timestamp(dump) do

integration_test/pg/storage_test.exs

@@ -59,6 +59,18 @@ defmodule Ecto.Integration.StorageTest do
     System.cmd("psql", args, env: env)
   end
 
+  def create_schema(database, schema) do
+    run_psql(~s[CREATE SCHEMA #{schema}], [database])
+  end
+
+  def drop_schema(database, schema) do
+    run_psql(~s[DROP SCHEMA #{schema}], [database])
+  end
+
+  def drop_schema_migrations_table(database, schema) do
+    run_psql(~s[DROP TABLE #{schema}.schema_migrations], [database])
+  end
+
   test "storage up (twice in a row)" do
     assert Postgres.storage_up(params()) == :ok
     assert Postgres.storage_up(params()) == {:error, :already_up}
@@ -147,6 +159,72 @@ defmodule Ecto.Integration.StorageTest do
     assert contents =~ ~s[INSERT INTO public."schema_migrations" (version) VALUES]
   end
 
+  test "when :dump_prefixes is not provided, structure is dumped for all schemas but only public schema migration records are inserted" do
+    # Create the test_schema schema
+    create_schema(PoolRepo.config()[:database], "test_schema")
+
+    # Run migrations
+    version = @base_migration + System.unique_integer([:positive])
+    :ok = Ecto.Migrator.up(PoolRepo, version, Migration, log: false)
+    :ok = Ecto.Migrator.up(PoolRepo, version, Migration, log: false, prefix: "test_schema")
+
+    {:ok, path} = Postgres.structure_dump(tmp_path(), TestRepo.config())
+    contents = File.read!(path)
+
+    assert contents =~ "CREATE TABLE public.schema_migrations"
+    assert contents =~ ~s[INSERT INTO public."schema_migrations" (version) VALUES (#{version})]
+    assert contents =~ "CREATE TABLE test_schema.schema_migrations"
+    refute contents =~ ~s[INSERT INTO test_schema."schema_migrations" (version) VALUES (#{version})]
+  after
+    drop_schema_migrations_table(PoolRepo.config()[:database], "test_schema")
+    drop_schema(PoolRepo.config()[:database], "test_schema")
+  end
+
+  test "dumps structure and schema_migration records from multiple schemas" do
+    # Create the test_schema schema
+    create_schema(PoolRepo.config()[:database], "test_schema")
+
+    # Run migrations
+    version = @base_migration + System.unique_integer([:positive])
+    :ok = Ecto.Migrator.up(PoolRepo, version, Migration, log: false)
+    :ok = Ecto.Migrator.up(PoolRepo, version, Migration, log: false, prefix: "test_schema")
+
+    config = Keyword.put(TestRepo.config(), :dump_prefixes, ["public", "test_schema"])
+    {:ok, path} = Postgres.structure_dump(tmp_path(), config)
+    contents = File.read!(path)
+
+    assert contents =~ "CREATE TABLE public.schema_migrations"
+    assert contents =~ ~s[INSERT INTO public."schema_migrations" (version) VALUES (#{version})]
+    assert contents =~ "CREATE TABLE test_schema.schema_migrations"
+    assert contents =~ ~s[INSERT INTO test_schema."schema_migrations" (version) VALUES (#{version})]
+  after
+    drop_schema_migrations_table(PoolRepo.config()[:database], "test_schema")
+    drop_schema(PoolRepo.config()[:database], "test_schema")
+  end
+
+  test "dumps structure and schema_migration records only from queried schema" do
+    # Create the test_schema schema
+    create_schema(PoolRepo.config()[:database], "test_schema")
+
+    # Run migrations
+    version = @base_migration + System.unique_integer([:positive])
+    :ok = Ecto.Migrator.up(PoolRepo, version, Migration, log: false)
+    :ok = Ecto.Migrator.up(PoolRepo, version, Migration, log: false, prefix: "test_schema")
+
+    config = Keyword.put(TestRepo.config(), :dump_prefixes, ["test_schema"])
+    {:ok, path} = Postgres.structure_dump(tmp_path(), config)
+    contents = File.read!(path)
+
+    refute contents =~ "CREATE TABLE public.schema_migrations"
+    refute contents =~ ~s[INSERT INTO public."schema_migrations" (version) VALUES (#{version})]
+    assert contents =~ "CREATE TABLE test_schema.schema_migrations"
+    assert contents =~ ~s[INSERT INTO test_schema."schema_migrations" (version) VALUES (#{version})]
+  after
+    drop_schema_migrations_table(PoolRepo.config()[:database], "test_schema")
+    drop_schema(PoolRepo.config()[:database], "test_schema")
+  end
+
+
   test "storage status is up when database is created" do
     create_database()
     assert :up == Postgres.storage_status(params())

lib/ecto/adapters/myxql.ex

@@ -50,6 +50,10 @@ defmodule Ecto.Adapters.MyXQL do
     * `:charset` - the database encoding (default: "utf8mb4")
     * `:collation` - the collation order
     * `:dump_path` - where to place dumped structures
+    * `:dump_prefixes` - list of prefixes that will be included in the
+      structure dump. When specified, the prefixes will have their definitions
+      dumped along with the data in their migration table. When it is not
+      specified, only the configured database and its migration table are dumped.
 
   ### After connect callback
 
@@ -312,27 +316,38 @@ defmodule Ecto.Adapters.MyXQL do
   def structure_dump(default, config) do
     table = config[:migration_source] || "schema_migrations"
     path  = config[:dump_path] || Path.join(default, "structure.sql")
+    prefixes = config[:dump_prefixes] || [config[:database]]
 
-    with {:ok, versions} <- select_versions(table, config),
-         {:ok, contents} <- mysql_dump(config),
+    with {:ok, versions} <- select_versions(prefixes, table, config),
+         {:ok, contents} <- mysql_dump(prefixes, config),
          {:ok, contents} <- append_versions(table, versions, contents) do
       File.mkdir_p!(Path.dirname(path))
       File.write!(path, contents)
       {:ok, path}
     end
   end
 
-  defp select_versions(table, config) do
-    case run_query(~s[SELECT version FROM `#{table}` ORDER BY version], config) do
-      {:ok, %{rows: rows}} -> {:ok, Enum.map(rows, &hd/1)}
-      {:error, %{mysql: %{name: :ER_NO_SUCH_TABLE}}} -> {:ok, []}
+  defp select_versions(prefixes, table, config) do
+    result =
+      Enum.reduce_while(prefixes, [], fn prefix, versions ->
+        case run_query(~s[SELECT version FROM `#{prefix}`.`#{table}` ORDER BY version], config) do
+          {:ok, %{rows: rows}} -> {:cont, Enum.map(rows, &{prefix, hd(&1)}) ++ versions}
+          {:error, %{mysql: %{name: :ER_NO_SUCH_TABLE}}} -> {:cont, versions}
+          {:error, _} = error -> {:halt, error}
+          {:exit, exit} -> {:halt, {:error, exit_to_exception(exit)}}
+        end
+      end)
+
+    case result do
       {:error, _} = error -> error
-      {:exit, exit} -> {:error, exit_to_exception(exit)}
+      versions -> {:ok, versions}
     end
   end
 
-  defp mysql_dump(config) do
-    case run_with_cmd("mysqldump", config, ["--no-data", "--routines", config[:database]]) do
+  defp mysql_dump(prefixes, config) do
+    args = ["--no-data", "--routines", "--databases" | prefixes]
+
+    case run_with_cmd("mysqldump", config, args) do
       {output, 0} -> {:ok, output}
       {output, _} -> {:error, output}
     end
@@ -341,10 +356,14 @@ defmodule Ecto.Adapters.MyXQL do
   defp append_versions(_table, [], contents) do
     {:ok, contents}
   end
+
   defp append_versions(table, versions, contents) do
-    {:ok,
-      contents <>
-      Enum.map_join(versions, &~s[INSERT INTO `#{table}` (version) VALUES (#{&1});\n])}
+    sql_statements =
+      Enum.map_join(versions, fn {prefix, version} ->
+        ~s[INSERT INTO `#{prefix}`.`#{table}` (version) VALUES (#{version});\n]
+      end)
+
+    {:ok, contents <> sql_statements}
   end
 
   @impl true

lib/ecto/adapters/postgres.ex

@@ -93,6 +93,11 @@ defmodule Ecto.Adapters.Postgres do
     * `:lc_collate` - the collation order
     * `:lc_ctype` - the character classification
     * `:dump_path` - where to place dumped structures
+    * `dump_prefixes` - list of prefixes that will be included in the structure dump.
+      When specified, the prefixes will have their definitions dumped along with the
+      data in their migration table. When it is not specified, the configured
+      database has the definitions dumped from all of its schemas but only
+      the data from the migration table from the `public` schema is included.
     * `:force_drop` - force the database to be dropped even
       if it has connections to it (requires PostgreSQL 13+)
 
@@ -344,18 +349,36 @@ defmodule Ecto.Adapters.Postgres do
   end
 
   defp select_versions(table, config) do
-    case run_query(~s[SELECT version FROM public."#{table}" ORDER BY version], config) do
-      {:ok, %{rows: rows}} -> {:ok, Enum.map(rows, &hd/1)}
-      {:error, %{postgres: %{code: :undefined_table}}} -> {:ok, []}
+    prefixes = config[:dump_prefixes] || ["public"]
+
+    result =
+      Enum.reduce_while(prefixes, [], fn prefix, versions ->
+        case run_query(~s[SELECT version FROM #{prefix}."#{table}" ORDER BY version], config) do
+          {:ok, %{rows: rows}} -> {:cont, Enum.map(rows, &{prefix, hd(&1)}) ++ versions }
+          {:error, %{postgres: %{code: :undefined_table}}} -> {:cont, versions}
+          {:error, _} = error -> {:halt, error}
+        end
+      end)
+
+    case result do
       {:error, _} = error -> error
+      versions -> {:ok, versions}
     end
   end
 
   defp pg_dump(default, config) do
     path = config[:dump_path] || Path.join(default, "structure.sql")
+    prefixes = config[:dump_prefixes] || []
+    non_prefix_args = ["--file", path, "--schema-only", "--no-acl", "--no-owner"]
+
+    args =
+      Enum.reduce(prefixes, non_prefix_args, fn prefix, acc ->
+        ["-n", prefix | acc]
+      end)
+
     File.mkdir_p!(Path.dirname(path))
 
-    case run_with_cmd("pg_dump", config, ["--file", path, "--schema-only", "--no-acl", "--no-owner"]) do
+    case run_with_cmd("pg_dump", config, args) do
       {_output, 0} ->
         {:ok, path}
       {output, _} ->
@@ -368,7 +391,10 @@ defmodule Ecto.Adapters.Postgres do
   end
 
   defp append_versions(table, versions, path) do
-    sql = Enum.map_join(versions, &~s[INSERT INTO public."#{table}" (version) VALUES (#{&1});\n])
+    sql =
+      Enum.map_join(versions, fn {prefix, version} ->
+        ~s[INSERT INTO #{prefix}."#{table}" (version) VALUES (#{version});\n]
+      end)
 
     File.open!(path, [:append], fn file ->
       IO.write(file, sql)

lib/mix/tasks/ecto.dump.ex

@@ -17,7 +17,8 @@ defmodule Mix.Tasks.Ecto.Dump do
     quiet: :boolean,
     repo: [:string, :keep],
     no_compile: :boolean,
-    no_deps_check: :boolean
+    no_deps_check: :boolean,
+    prefix: [:string, :keep]
   ]
 
   @moduledoc """
@@ -46,12 +47,30 @@ defmodule Mix.Tasks.Ecto.Dump do
     * `-q`, `--quiet` - run the command quietly
     * `--no-compile` - does not compile applications before dumping
     * `--no-deps-check` - does not check dependencies before dumping
+    * `--prefix` - prefix that will be included in the structure dump.
+      Can include multiple prefixes (ex. --prefix foo --prefix bar).
+      When specified, the prefixes will have their definitions dumped along
+      with the data in their migration table. The default behavior is
+      dependent on the adapter for backwards compatibility reasons.
+      For PostgreSQL, the configured database has the definitions dumped
+      from all of its schemas but only the data from the migration table
+      from the `public` schema is included. For MySQL, only the configured
+      database and its migration table are dumped.
   """
 
   @impl true
   def run(args) do
     {opts, _} = OptionParser.parse! args, strict: @switches, aliases: @aliases
-    opts = Keyword.merge(@default_opts, opts)
+
+    dump_prefixes =
+      case Keyword.get_values(opts, :prefix) do
+        [_ | _] = prefixes -> prefixes
+        [] -> nil
+      end
+
+    opts = @default_opts
+    |> Keyword.merge(opts)
+    |> Keyword.put(:dump_prefixes, dump_prefixes)
 
     Enum.each parse_repo(args), fn repo ->
       ensure_repo(repo, args)

test/ecto/adapters/postgres_test.exs

@@ -55,18 +55,18 @@ defmodule Ecto.Adapters.PostgresTest do
   defp delete_all(query), do: query |> SQL.delete_all |> IO.iodata_to_binary()
   defp execute_ddl(query), do: query |> SQL.execute_ddl |> Enum.map(&IO.iodata_to_binary/1)
 
-  defp insert(prefx, table, header, rows, on_conflict, returning, placeholders \\ []) do
+  defp insert(prefix, table, header, rows, on_conflict, returning, placeholders \\ []) do
     IO.iodata_to_binary(
-      SQL.insert(prefx, table, header, rows, on_conflict, returning, placeholders)
+      SQL.insert(prefix, table, header, rows, on_conflict, returning, placeholders)
     )
   end
 
-  defp update(prefx, table, fields, filter, returning) do
-    IO.iodata_to_binary(SQL.update(prefx, table, fields, filter, returning))
+  defp update(prefix, table, fields, filter, returning) do
+    IO.iodata_to_binary(SQL.update(prefix, table, fields, filter, returning))
   end
 
-  defp delete(prefx, table, filter, returning) do
-    IO.iodata_to_binary(SQL.delete(prefx, table, filter, returning))
+  defp delete(prefix, table, filter, returning) do
+    IO.iodata_to_binary(SQL.delete(prefix, table, filter, returning))
   end
 
   test "from" do