use input redirection for mysql structure_load

Author: greg-rychlewski

Earthfile

@@ -92,7 +92,8 @@ integration-test-mysql:
             # the default authentication plugin for MySQL 8 is sha 256 but it doesn't come with the docker image. falling back to the 5.7 way
             --default-authentication-plugin=mysql_native_password; \
             # wait for mysql to start
-            while ! mysqladmin ping --host=127.0.0.1 --port=3306 --protocol=TCP --silent; do \
+            while ! mysqladmin ping --host=127.0.0.1 --port=3306 --protocol=TCP; do \
+                docker logs mysql; \
                 test "$(date +%s)" -le "$timeout" || (echo "timed out waiting for mysql"; exit 1); \
                 echo "waiting for mysql"; \
                 sleep 1; \

lib/ecto/adapters/myxql.ex

@@ -419,13 +419,15 @@ defmodule Ecto.Adapters.MyXQL do
     case File.read(path) do
       {:ok, contents} ->
         args = [
-          "--execute",
-          "SET FOREIGN_KEY_CHECKS = 0; " <> contents <> "; SET FOREIGN_KEY_CHECKS = 1",
+          "--silent",
+          "--batch",
+          "--unbuffered",
+          "--init-command=SET FOREIGN_KEY_CHECKS = 0;",
           "--database",
           config[:database]
         ]
 
-        case run_with_cmd("mysql", config, args) do
+        case run_with_port("mysql", config, args, contents) do
           {_output, 0} -> {:ok, path}
           {output, _} -> {:error, output}
         end
@@ -497,6 +499,12 @@ defmodule Ecto.Adapters.MyXQL do
               "please guarantee it is available before running ecto commands"
     end
 
+    {args, cmd_opts} = args_cmd_opts(opts, opt_args, cmd_opts)
+
+    System.cmd(cmd, args, cmd_opts)
+  end
+
+  defp args_cmd_opts(opts, opt_args, cmd_opts) do
     env =
       if password = opts[:password] do
         [{"MYSQL_PWD", password}]
@@ -530,6 +538,80 @@ defmodule Ecto.Adapters.MyXQL do
       |> Keyword.put_new(:stderr_to_stdout, true)
       |> Keyword.update(:env, env, &Enum.concat(env, &1))
 
-    System.cmd(cmd, args, cmd_opts)
+    {args, cmd_opts}
+  end
+
+  # Ported from Elixir System.cmd implementation with the
+  # intent of using file redirection for passing dumps
+  # into the mysql client so that users don't run into
+  # shell limits when files are too large
+  defp run_with_port(cmd, opts, opt_args, contents, cmd_opts \\ []) do
+    abs_cmd = System.find_executable(cmd)
+
+    unless abs_cmd do
+      raise "could not find executable `#{cmd}` in path, " <>
+              "please guarantee it is available before running ecto commands"
+    end
+
+    abs_cmd = String.to_charlist(abs_cmd)
+    {args, cmd_opts} = args_cmd_opts(opts, opt_args, cmd_opts)
+
+    port_opts = port_opts(cmd_opts, args: args)
+    port = Port.open({:spawn_executable, abs_cmd}, port_opts)
+    Port.command(port, contents)
+    # Use this as a signal to close the port since we cannot
+    # send an exit command to mysql in batch mode
+    Port.command(port, ";SELECT '__ECTO_EOF__';\n")
+
+    {initial, fun} = Collectable.into("")
+
+    try do
+      collect_output(port, initial, fun)
+    catch
+      kind, reason ->
+        fun.(initial, :halt)
+        :erlang.raise(kind, reason, __STACKTRACE__)
+    else
+      {acc, status} -> {fun.(acc, :done), status}
+    end
+  end
+
+  defp port_opts([{:stderr_to_stdout, true} | t], acc),
+    do: port_opts(t, [:stderr_to_stdout | acc])
+
+  defp port_opts([{:stderr_to_stdout, _} | t], acc),
+    do: port_opts(t, acc)
+
+  defp port_opts([{:env, enum} | t], acc),
+    do: port_opts(t, [{:env, validate_env(enum)} | acc])
+
+  defp port_opts([], acc) do
+    [:use_stdio, :exit_status, :binary, :hide] ++ acc
+  end
+
+  defp validate_env(enum) do
+    Enum.map(enum, fn
+      {k, nil} ->
+        {String.to_charlist(k), false}
+
+      {k, v} ->
+        {String.to_charlist(k), String.to_charlist(v)}
+
+      other ->
+        raise ArgumentError, "invalid environment key-value #{inspect(other)}"
+    end)
+  end
+
+  defp collect_output(port, acc, fun) do
+    receive do
+      {^port, {:data, "__ECTO_EOF__" <> _rest}} ->
+        {acc, 0}
+
+      {^port, {:data, data}} ->
+        collect_output(port, fun.(acc, {:cont, data}), fun)
+
+      {^port, {:exit_status, status}} ->
+        {acc, status}
+    end
   end
 end