make Ch.stream collectable

[ruslandoga]

This PR makes Ch.Stream.t collectable, meaning it can be Enum.intoed or Stream.intoed with iodata and that iodata will be sent as an HTTP chunk.

Example from plausible/analytics#3692:

  @doc """
  Creates a collectable stream which pipes the data to an `imported_*` table.

  Example usage:

      {:ok, pool} = Ch.start_link(pool_size: 1)

      DBConnection.run(pool, fn conn ->
        File.stream!("imported_browsers.csv", _64kb = 64000)
        |> Stream.into(import_stream(conn, _site_id = 12, _table = "imported_browsers", _format = "CSVWithNames"))
        |> Stream.run()
      end)

  """
  @spec import_stream(DBConnection.t(), pos_integer, String.t(), String.t(), [Ch.query_option()]) ::
          Ch.Stream.t()
  def import_stream(conn, site_id, table, format, opts \\ []) do
    :ok = ensure_supported_format(format)

    statement =
      [
        "INSERT INTO {table:Identifier} SELECT {site_id:UInt64} AS site_id, * FROM input('",
        input_structure(table),
        "') FORMAT ",
        format
      ]

    Ch.stream(conn, statement, %{"table" => table, "site_id" => site_id}, opts)
  end