Implement multiple statements and multiple result sets.

This also makes it possible to call prepared statements.

README.md

@@ -60,6 +60,10 @@ case Result of
         io:format("Inserted 0 rows.~n")
 end
 
+%% Multiple queries and multiple result sets
+{ok, [{[<<"foo">>], [[42]]}, {[<<"bar">>], [[<<"baz">>]]}]} =
+    mysql:query(Pid, "SELECT 42 AS foo; SELECT 'baz' AS bar;"),
+
 %% Graceful timeout handling: SLEEP() returns 1 when interrupted
 {ok, [<<"SLEEP(5)">>], [[1]]} =
     mysql:query(Pid, <<"SELECT SLEEP(5)">>, 1000),

include/protocol.hrl

@@ -30,28 +30,33 @@
 %% Client: Handshake Response Packet contains a schema-name
 -define(CLIENT_CONNECT_WITH_DB, 16#00000008).
 
-%% Server: supports the 4.1 protocol 
-%% Client: uses the 4.1 protocol 
+%% Server: supports the 4.1 protocol
+%% Client: uses the 4.1 protocol
 -define(CLIENT_PROTOCOL_41, 16#00000200).
 
-%% Server: can send status flags in EOF_Packet 
-%% Client: expects status flags in EOF_Packet 
+%% Server: can send status flags in EOF_Packet
+%% Client: expects status flags in EOF_Packet
 -define(CLIENT_TRANSACTIONS, 16#00002000).
 
-%% Server: supports Authentication::Native41 
-%% Client: supports Authentication::Native41 
+%% Server: supports Authentication::Native41
+%% Client: supports Authentication::Native41
 -define(CLIENT_SECURE_CONNECTION, 16#00008000).
 
-%% Server: can handle multiple statements per COM_QUERY and COM_STMT_PREPARE 
-%% Client: may send multiple statements per COM_QUERY and COM_STMT_PREPARE 
+%% Server: can handle multiple statements per COM_QUERY and COM_STMT_PREPARE
+%% Client: may send multiple statements per COM_QUERY and COM_STMT_PREPARE
 %% Requires: CLIENT_PROTOCOL_41
 -define(CLIENT_MULTI_STATEMENTS, 16#00010000).
 
-%% Server: can send multiple resultsets for COM_QUERY 
-%% Client: can handle multiple resultsets for COM_QUERY 
+%% Server: can send multiple resultsets for COM_QUERY
+%% Client: can handle multiple resultsets for COM_QUERY
 %% Requires: CLIENT_PROTOCOL_41
 -define(CLIENT_MULTI_RESULTS, 16#00020000).
 
+%% Server: can send multiple resultsets for COM_STMT_EXECUTE
+%% Client: can handle multiple resultsets for COM_STMT_EXECUTE
+%% Requires: CLIENT_PROTOCOL_41
+-define(CLIENT_PS_MULTI_RESULTS, 16#00040000).
+
 %% Server: sends extra data in Initial Handshake Packet and supports the
 %%         pluggable authentication protocol.
 %% Client: supports auth plugins

include/records.hrl

@@ -52,5 +52,6 @@
 
 %% Response of a successfull prepare call.
 -record(prepared, {statement_id :: integer(),
+                   orig_query :: iodata(),
                    param_count :: integer(),
                    warning_count :: integer()}).

src/mysql.erl

@@ -29,7 +29,7 @@
          encode/2, in_transaction/1,
          transaction/2, transaction/3, transaction/4]).
 
--export_type([connection/0, server_reason/0]).
+-export_type([connection/0, server_reason/0, query_result/0]).
 
 -behaviour(gen_server).
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
@@ -61,6 +61,14 @@
 -type server_reason() :: {Code :: integer(), SQLState :: binary(),
                           Message :: binary()}.
 
+-type column_names() :: [binary()].
+-type rows() :: [[term()]].
+
+-type query_result() :: ok
+                      | {ok, column_names(), rows()}
+                      | {ok, [{column_names(), rows()}, ...]}
+                      | {error, server_reason()}.
+
 %% @doc Starts a connection gen_server process and connects to a database. To
 %% disconnect just do `exit(Pid, normal)'.
 %%
@@ -143,7 +151,8 @@ start_link(Options) ->
             lists:foreach(fun (Query) ->
                               case mysql:query(Pid, Query) of
                                   ok -> ok;
-                                  {ok, _, _} -> ok
+                                  {ok, _, _} -> ok;
+                                  {ok, _} -> ok
                               end
                           end,
                           Queries),
@@ -158,12 +167,19 @@ start_link(Options) ->
     Ret.
 
 %% @doc Executes a query with the query timeout as given to start_link/1.
--spec query(Conn, Query) -> ok | {ok, ColumnNames, Rows} | {error, Reason}
+%%
+%% It is possible to execute multiple semicolon-separated queries.
+%%
+%% Results are returned in the form `{ok, ColumnNames, Rows}' if there is one
+%% result set. If there are more than one result sets, they are returned in the
+%% form `{ok, [{ColumnNames, Rows}, ...]}'.
+%%
+%% For queries that don't return any rows (INSERT, UPDATE, etc.) only the atom
+%% `ok' is returned.
+-spec query(Conn, Query) -> Result
     when Conn :: connection(),
          Query :: iodata(),
-         ColumnNames :: [binary()],
-         Rows :: [[term()]],
-         Reason :: server_reason().
+         Result :: query_result().
 query(Conn, Query) ->
     query_call(Conn, {query, Query}).
 
@@ -174,17 +190,17 @@ query(Conn, Query) ->
 %%
 %% If the 3rd argument is a timeout, it executes a plain query with this
 %% timeout.
+%%
+%% The return value is the same as for query/2.
+%%
 %% @see query/2.
 %% @see query/4.
--spec query(Conn, Query, Params | Timeout) -> ok | {ok, ColumnNames, Rows} |
-                                              {error, Reason}
+-spec query(Conn, Query, Params | Timeout) -> Result
     when Conn :: connection(),
          Query :: iodata(),
          Timeout :: timeout(),
          Params :: [term()],
-         ColumnNames :: [binary()],
-         Rows :: [[term()]],
-         Reason :: server_reason().
+         Result :: query_result().
 query(Conn, Query, Params) when is_list(Params) ->
     query_call(Conn, {param_query, Query, Params});
 query(Conn, Query, Timeout) when is_integer(Timeout); Timeout == infinity ->
@@ -198,15 +214,14 @@ query(Conn, Query, Timeout) when is_integer(Timeout); Timeout == infinity ->
 %%
 %% The minimum time the prepared statement is cached can be specified using the
 %% option `{query_cache_time, Milliseconds}' to start_link/1.
--spec query(Conn, Query, Params, Timeout) -> ok | {ok, ColumnNames, Rows} |
-                                             {error, Reason}
+%%
+%% The return value is the same as for query/2.
+-spec query(Conn, Query, Params, Timeout) -> Result
     when Conn :: connection(),
          Query :: iodata(),
          Timeout :: timeout(),
          Params :: [term()],
-         ColumnNames :: [binary()],
-         Rows :: [[term()]],
-         Reason :: server_reason().
+         Result :: query_result().
 query(Conn, Query, Params, Timeout) ->
     query_call(Conn, {param_query, Query, Params, Timeout}).
 
@@ -214,14 +229,11 @@ query(Conn, Query, Params, Timeout) ->
 %% to start_link/1.
 %% @see prepare/2
 %% @see prepare/3
--spec execute(Conn, StatementRef, Params) ->
-    ok | {ok, ColumnNames, Rows} | {error, Reason}
+-spec execute(Conn, StatementRef, Params) -> Result | {error, not_prepared}
   when Conn :: connection(),
        StatementRef :: atom() | integer(),
        Params :: [term()],
-       ColumnNames :: [binary()],
-       Rows :: [[term()]],
-       Reason :: server_reason() | not_prepared.
+       Result :: query_result().
 execute(Conn, StatementRef, Params) ->
     query_call(Conn, {execute, StatementRef, Params}).
 
@@ -229,14 +241,12 @@ execute(Conn, StatementRef, Params) ->
 %% @see prepare/2
 %% @see prepare/3
 -spec execute(Conn, StatementRef, Params, Timeout) ->
-    ok | {ok, ColumnNames, Rows} | {error, Reason}
+    Result | {error, not_prepared}
   when Conn :: connection(),
        StatementRef :: atom() | integer(),
        Params :: [term()],
        Timeout :: timeout(),
-       ColumnNames :: [binary()],
-       Rows :: [[term()]],
-       Reason :: server_reason() | not_prepared.
+       Result :: query_result().
 execute(Conn, StatementRef, Params, Timeout) ->
     query_call(Conn, {execute, StatementRef, Params, Timeout}).
 
@@ -422,7 +432,7 @@ transaction(Conn, Fun, Args, Retries) when is_list(Args),
 %% parametrized queries with placeholders.
 %%
 %% @see query/3
-%% @see execute/30
+%% @see execute/3
 -spec encode(connection(), term()) -> iodata().
 encode(Conn, Term) ->
     Term1 = case (is_list(Term) orelse is_binary(Term)) andalso
@@ -514,7 +524,9 @@ init(Opts) ->
 %%   <dt>`ok'</dt>
 %%   <dd>Success without returning any table data (UPDATE, etc.)</dd>
 %%   <dt>`{ok, ColumnNames, Rows}'</dt>
-%%   <dd>Queries returning table data</dd>
+%%   <dd>Queries returning one result set of table data</dd>
+%%   <dt>`{ok, [{ColumnNames, Rows}, ...]}'</dt>
+%%   <dd>Queries returning more than one result set of table data</dd>
 %%   <dt>`{error, ServerReason}'</dt>
 %%   <dd>MySQL server error</dd>
 %%   <dt>`{implicit_commit, NestingLevel, Query}'</dt>
@@ -541,7 +553,7 @@ handle_call({query, Query}, From, State) ->
     handle_call({query, Query, State#state.query_timeout}, From, State);
 handle_call({query, Query, Timeout}, _From, State) ->
     Socket = State#state.socket,
-    Rec = case mysql_protocol:query(Query, gen_tcp, Socket, Timeout) of
+    {ok, Recs} = case mysql_protocol:query(Query, gen_tcp, Socket, Timeout) of
         {error, timeout} when State#state.server_version >= [5, 0, 0] ->
             kill_query(State),
             mysql_protocol:fetch_query_response(gen_tcp, Socket, ?cmd_timeout);
@@ -552,32 +564,10 @@ handle_call({query, Query, Timeout}, _From, State) ->
         QueryResult ->
             QueryResult
     end,
-    State1 = update_state(State, Rec),
+    State1 = lists:foldl(fun update_state/2, State, Recs),
     State1#state.warning_count > 0 andalso State1#state.log_warnings
         andalso log_warnings(State1, Query),
-    case Rec of
-        #ok{status = Status} when Status band ?SERVER_STATUS_IN_TRANS == 0,
-                                  State1#state.transaction_level > 0 ->
-            %% DDL statements (e.g. CREATE TABLE, ALTER TABLE, etc.) result in
-            %% an implicit commit.
-            Reply = {implicit_commit, State1#state.transaction_level, Query},
-            {reply, Reply, State1#state{transaction_level = 0}};
-        #ok{} ->
-            {reply, ok, State1};
-        #resultset{cols = ColDefs, rows = Rows} ->
-            Names = [Def#col.name || Def <- ColDefs],
-            {reply, {ok, Names, Rows}, State1};
-        #error{code = Code} when State1#state.transaction_level > 0,
-                                 (Code == ?ERROR_DEADLOCK orelse
-                                  Code == ?ERROR_LOCK_WAIT_TIMEOUT) ->
-            %% These errors result in an implicit rollback.
-            Reply = {implicit_rollback, State1#state.transaction_level,
-                     error_to_reason(Rec)},
-            State2 = clear_transaction_status(State1),
-            {reply, Reply, State2};
-        #error{} ->
-            {reply, {error, error_to_reason(Rec)}, State1}
-    end;
+    handle_query_call_reply(Recs, Query, State1, []);
 handle_call({param_query, Query, Params}, From, State) ->
     handle_call({param_query, Query, Params, State#state.query_timeout}, From,
                 State);
@@ -593,7 +583,7 @@ handle_call({param_query, Query, Params, Timeout}, _From, State) ->
         not_found ->
             %% Prepare
             Rec = mysql_protocol:prepare(Query, gen_tcp, Socket),
-            %State1 = update_state(State, Rec),
+            %State1 = update_state(Rec, State),
             case Rec of
                 #error{} = E ->
                     {{error, error_to_reason(E)}, Cache};
@@ -609,10 +599,7 @@ handle_call({param_query, Query, Params, Timeout}, _From, State) ->
     case StmtResult of
         {ok, StmtRec} ->
             State1 = State#state{query_cache = Cache1},
-            {Reply, State2} = execute_stmt(StmtRec, Params, Timeout, State1),
-            State2#state.warning_count > 0 andalso State2#state.log_warnings
-                andalso log_warnings(State2, Query),
-            {reply, Reply, State2};
+            execute_stmt(StmtRec, Params, Timeout, State1);
         PrepareError ->
             {reply, PrepareError, State}
     end;
@@ -621,19 +608,14 @@ handle_call({execute, Stmt, Args}, From, State) ->
 handle_call({execute, Stmt, Args, Timeout}, _From, State) ->
     case dict:find(Stmt, State#state.stmts) of
         {ok, StmtRec} ->
-            {Reply, State1} = execute_stmt(StmtRec, Args, Timeout, State),
-            State1#state.warning_count > 0 andalso State1#state.log_warnings
-                andalso log_warnings(State1,
-                                     io_lib:format("prepared statement ~p",
-                                                   [Stmt])),
-            {reply, Reply, State1};
+            execute_stmt(StmtRec, Args, Timeout, State);
         error ->
             {reply, {error, not_prepared}, State}
     end;
 handle_call({prepare, Query}, _From, State) ->
     #state{socket = Socket} = State,
     Rec = mysql_protocol:prepare(Query, gen_tcp, Socket),
-    State1 = update_state(State, Rec),
+    State1 = update_state(Rec, State),
     case Rec of
         #error{} = E ->
             {reply, {error, error_to_reason(E)}, State1};
@@ -653,7 +635,7 @@ handle_call({prepare, Name, Query}, _From, State) when is_atom(Name) ->
             State
     end,
     Rec = mysql_protocol:prepare(Query, gen_tcp, Socket),
-    State2 = update_state(State1, Rec),
+    State2 = update_state(Rec, State1),
     case Rec of
         #error{} = E ->
             {reply, {error, error_to_reason(E)}, State2};
@@ -695,8 +677,9 @@ handle_call(start_transaction, _From,
         0 -> <<"BEGIN">>;
         _ -> <<"SAVEPOINT s", (integer_to_binary(L))/binary>>
     end,
-    Res = #ok{} = mysql_protocol:query(Query, gen_tcp, Socket, ?cmd_timeout),
-    State1 = update_state(State, Res),
+    {ok, [Res = #ok{}]} = mysql_protocol:query(Query, gen_tcp, Socket,
+                                               ?cmd_timeout),
+    State1 = update_state(Res, State),
     {reply, ok, State1#state{transaction_level = L + 1}};
 handle_call(rollback, _From, State = #state{socket = Socket, status = Status,
                                             transaction_level = L})
@@ -705,8 +688,9 @@ handle_call(rollback, _From, State = #state{socket = Socket, status = Status,
         1 -> <<"ROLLBACK">>;
         _ -> <<"ROLLBACK TO s", (integer_to_binary(L - 1))/binary>>
     end,
-    Res = #ok{} = mysql_protocol:query(Query, gen_tcp, Socket, ?cmd_timeout),
-    State1 = update_state(State, Res),
+    {ok, [Res = #ok{}]} = mysql_protocol:query(Query, gen_tcp, Socket,
+                                               ?cmd_timeout),
+    State1 = update_state(Res, State),
     {reply, ok, State1#state{transaction_level = L - 1}};
 handle_call(commit, _From, State = #state{socket = Socket, status = Status,
                                           transaction_level = L})
@@ -715,8 +699,9 @@ handle_call(commit, _From, State = #state{socket = Socket, status = Status,
         1 -> <<"COMMIT">>;
         _ -> <<"RELEASE SAVEPOINT s", (integer_to_binary(L - 1))/binary>>
     end,
-    Res = #ok{} = mysql_protocol:query(Query, gen_tcp, Socket, ?cmd_timeout),
-    State1 = update_state(State, Res),
+    {ok, [Res = #ok{}]} = mysql_protocol:query(Query, gen_tcp, Socket,
+                                               ?cmd_timeout),
+    State1 = update_state(Res, State),
     {reply, ok, State1#state{transaction_level = L - 1}}.
 
 %% @private
@@ -740,7 +725,7 @@ handle_info(query_cache, State = #state{query_cache = Cache,
     {noreply, State#state{query_cache = Cache1}};
 handle_info(ping, State) ->
     Ok = mysql_protocol:ping(gen_tcp, State#state.socket),
-    {noreply, update_state(State, Ok)};
+    {noreply, update_state(Ok, State)};
 handle_info(_Info, State) ->
     {noreply, State}.
 
@@ -775,7 +760,8 @@ query_call(Conn, CallReq) ->
 
 %% @doc Executes a prepared statement and returns {Reply, NextState}.
 execute_stmt(Stmt, Args, Timeout, State = #state{socket = Socket}) ->
-    Rec = case mysql_protocol:execute(Stmt, Args, gen_tcp, Socket, Timeout) of
+    {ok, Recs} = case mysql_protocol:execute(Stmt, Args, gen_tcp, Socket,
+                                             Timeout) of
         {error, timeout} when State#state.server_version >= [5, 0, 0] ->
             kill_query(State),
             mysql_protocol:fetch_execute_response(gen_tcp, Socket,
@@ -787,24 +773,10 @@ execute_stmt(Stmt, Args, Timeout, State = #state{socket = Socket}) ->
         QueryResult ->
             QueryResult
     end,
-    State1 = update_state(State, Rec),
-    case Rec of
-        #ok{} ->
-            {ok, State1};
-        #error{code = Code} when State1#state.transaction_level > 0,
-                                 (Code == ?ERROR_DEADLOCK orelse
-                                  Code == ?ERROR_LOCK_WAIT_TIMEOUT) ->
-            %% Implicit rollback.
-            Reply = {implicit_rollback, State1#state.transaction_level,
-                     error_to_reason(Rec)},
-            State2 = clear_transaction_status(State1),
-            {Reply, State2};
-        #error{} = E ->
-            {{error, error_to_reason(E)}, State1};
-        #resultset{cols = ColDefs, rows = Rows} ->
-            Names = [Def#col.name || Def <- ColDefs],
-            {{ok, Names, Rows}, State1}
-    end.
+    State1 = lists:foldl(fun update_state/2, State, Recs),
+    State1#state.warning_count > 0 andalso State1#state.log_warnings
+        andalso log_warnings(State1, Stmt#prepared.orig_query),
+    handle_query_call_reply(Recs, Stmt#prepared.orig_query, State1, []).
 
 %% @doc Produces a tuple to return as an error reason.
 -spec error_to_reason(#error{}) -> server_reason().
@@ -813,8 +785,8 @@ error_to_reason(#error{code = Code, state = State, msg = Msg}) ->
 
 %% @doc Updates a state with information from a response. Also re-schedules
 %% ping.
--spec update_state(#state{}, #ok{} | #eof{} | any()) -> #state{}.
-update_state(State, Rec) ->
+-spec update_state(#ok{} | #eof{} | any(), #state{}) -> #state{}.
+update_state(Rec, State) ->
     State1 = case Rec of
         #ok{status = S, affected_rows = R, insert_id = Id, warning_count = W} ->
             State#state{status = S, affected_rows = R, insert_id = Id,
@@ -830,6 +802,40 @@ update_state(State, Rec) ->
     end,
     schedule_ping(State1).
 
+%% @doc Produces a reply for handle_call/3 for queries and prepared statements.
+handle_query_call_reply([], _Query, State, ResultSetsAcc) ->
+    Reply = case ResultSetsAcc of
+        []                    -> ok;
+        [{ColumnNames, Rows}] -> {ok, ColumnNames, Rows};
+        [_|_]                 -> {ok, lists:reverse(ResultSetsAcc)}
+    end,
+    {reply, Reply, State};
+handle_query_call_reply([Rec|Recs], Query, State, ResultSetsAcc) ->
+    case Rec of
+        #ok{status = Status} when Status band ?SERVER_STATUS_IN_TRANS == 0,
+                                  State#state.transaction_level > 0 ->
+            %% DDL statements (e.g. CREATE TABLE, ALTER TABLE, etc.) result in
+            %% an implicit commit.
+            Reply = {implicit_commit, State#state.transaction_level, Query},
+            {reply, Reply, State#state{transaction_level = 0}};
+        #ok{} ->
+            handle_query_call_reply(Recs, Query, State, ResultSetsAcc);
+        #resultset{cols = ColDefs, rows = Rows} ->
+            Names = [Def#col.name || Def <- ColDefs],
+            ResultSetsAcc1 = [{Names, Rows} | ResultSetsAcc],
+            handle_query_call_reply(Recs, Query, State, ResultSetsAcc1);
+        #error{code = Code} when State#state.transaction_level > 0,
+                                 (Code == ?ERROR_DEADLOCK orelse
+                                  Code == ?ERROR_LOCK_WAIT_TIMEOUT) ->
+            %% These errors result in an implicit rollback.
+            Reply = {implicit_rollback, State#state.transaction_level,
+                     error_to_reason(Rec)},
+            State2 = clear_transaction_status(State),
+            {reply, Reply, State2};
+        #error{} ->
+            {reply, {error, error_to_reason(Rec)}, State}
+    end.
+
 %% @doc Schedules (or re-schedules) ping.
 schedule_ping(State = #state{ping_timeout = infinity}) ->
     State;
@@ -846,8 +852,9 @@ clear_transaction_status(State = #state{status = Status}) ->
 
 %% @doc Fetches and logs warnings. Query is the query that gave the warnings.
 log_warnings(#state{socket = Socket}, Query) ->
-    #resultset{rows = Rows} = mysql_protocol:query(<<"SHOW WARNINGS">>, gen_tcp,
-                                                   Socket, ?cmd_timeout),
+    {ok, [#resultset{rows = Rows}]} = mysql_protocol:query(<<"SHOW WARNINGS">>,
+                                                           gen_tcp, Socket,
+                                                           ?cmd_timeout),
     Lines = [[Level, " ", integer_to_binary(Code), ": ", Message, "\n"]
              || [Level, Code, Message] <- Rows],
     error_logger:warning_msg("~s in ~s~n", [Lines, Query]).
@@ -867,8 +874,9 @@ kill_query(#state{connection_id = ConnId, host = Host, port = Port,
         #handshake{} ->
             %% Kill and disconnect
             IdBin = integer_to_binary(ConnId),
-            #ok{} = mysql_protocol:query(<<"KILL QUERY ", IdBin/binary>>,
-                                         gen_tcp, Socket, ?cmd_timeout),
+            {ok, [#ok{}]} = mysql_protocol:query(<<"KILL QUERY ",
+                                                   IdBin/binary>>, gen_tcp,
+                                                 Socket, ?cmd_timeout),
             mysql_protocol:quit(gen_tcp, Socket);
         #error{} = E ->
             error_logger:error_msg("Failed to connect to kill query: ~p",

src/mysql_protocol.erl

@@ -35,6 +35,7 @@
 
 -include("records.hrl").
 -include("protocol.hrl").
+-include("server_status.hrl").
 
 %% Macros for pattern matching on packets.
 -define(ok_pattern, <<?OK, _/binary>>).
@@ -77,7 +78,7 @@ ping(TcpModule, Socket) ->
     parse_ok_packet(OkPacket).
 
 -spec query(Query :: iodata(), atom(), term(), timeout()) ->
-    #ok{} | #resultset{} | #error{} | {error, timeout}.
+    {ok, [#ok{} | #resultset{} | #error{}]} | {error, timeout}.
 query(Query, TcpModule, Socket, Timeout) ->
     Req = <<?COM_QUERY, (iolist_to_binary(Query))/binary>>,
     SeqNum0 = 0,
@@ -87,27 +88,7 @@ query(Query, TcpModule, Socket, Timeout) ->
 %% @doc This is used by query/4. If query/4 returns {error, timeout}, this
 %% function can be called to retry to fetch the results of the query. 
 fetch_query_response(TcpModule, Socket, Timeout) ->
-    case recv_packet(TcpModule, Socket, Timeout, any) of
-        {ok, ?ok_pattern = Ok, _} ->
-            parse_ok_packet(Ok);
-        {ok, ?error_pattern = Error, _} ->
-            parse_error_packet(Error);
-        {ok, ResultPacket, SeqNum2} ->
-            %% The first packet in a resultset is only the column count.
-            {ColumnCount, <<>>} = lenenc_int(ResultPacket),
-            case fetch_resultset(TcpModule, Socket, ColumnCount, SeqNum2) of
-                #error{} = E ->
-                    E;
-                #resultset{cols = ColDefs, rows = Rows} = R ->
-                    %% Parse the rows according to the 'text protocol'
-                    %% representation.
-                    Rows1 = [decode_text_row(ColumnCount, ColDefs, Row)
-                             || Row <- Rows],
-                    R#resultset{rows = Rows1}
-            end;
-        {error, timeout} ->
-            {error, timeout}
-    end.
+    fetch_response(TcpModule, Socket, Timeout, text, []).
 
 %% @doc Prepares a statement.
 -spec prepare(iodata(), atom(), term()) -> #error{} | #prepared{}.
@@ -139,6 +120,7 @@ prepare(Query, TcpModule, Socket) ->
                 fetch_column_definitions_if_any(NumColumns, TcpModule, Socket,
                                                 SeqNum3),
             #prepared{statement_id = StmtId,
+                      orig_query = Query,
                       param_count = NumParams,
                       warning_count = WarningCount}
     end.
@@ -152,7 +134,7 @@ unprepare(#prepared{statement_id = Id}, TcpModule, Socket) ->
 
 %% @doc Executes a prepared statement.
 -spec execute(#prepared{}, [term()], atom(), term(), timeout()) ->
-    #ok{} | #resultset{} | #error{} | {error, timeout}.
+    {ok, [#ok{} | #resultset{} | #error{}]} | {error, timeout}.
 execute(#prepared{statement_id = Id, param_count = ParamCount}, ParamValues,
         TcpModule, Socket, Timeout) when ParamCount == length(ParamValues) ->
     %% Flags Constant Name
@@ -186,28 +168,7 @@ execute(#prepared{statement_id = Id, param_count = ParamCount}, ParamValues,
 %% @doc This is used by execute/5. If execute/5 returns {error, timeout}, this
 %% function can be called to retry to fetch the results of the query.
 fetch_execute_response(TcpModule, Socket, Timeout) ->
-    case recv_packet(TcpModule, Socket, Timeout, any) of
-        {ok, ?ok_pattern = Ok, _} ->
-            parse_ok_packet(Ok);
-        {ok, ?error_pattern = Error, _} ->
-            parse_error_packet(Error);
-        {ok, ResultPacket, SeqNum2} ->
-            %% The first packet in a resultset is only the column count.
-            {ColumnCount, <<>>} = lenenc_int(ResultPacket),
-            case fetch_resultset(TcpModule, Socket, ColumnCount, SeqNum2) of
-                #error{} = E ->
-                    %% TODO: Find a way to get here and write a testcase.
-                    E;
-                #resultset{cols = ColDefs, rows = Rows} = R ->
-                    %% Parse the rows according to the 'binary protocol'
-                    %% representation.
-                    Rows1 = [decode_binary_row(ColumnCount, ColDefs, Row)
-                             || Row <- Rows],
-                    R#resultset{rows = Rows1}
-            end;
-        {error, timeout} ->
-            {error, timeout}
-    end.
+    fetch_response(TcpModule, Socket, Timeout, binary, []).
 
 %% --- internal ---
 
@@ -268,7 +229,10 @@ build_handshake_response(Handshake, Username, Password, Database) ->
     %% We require these capabilities. Make sure the server handles them.
     CapabilityFlags0 = ?CLIENT_PROTOCOL_41 bor
                        ?CLIENT_TRANSACTIONS bor
-                       ?CLIENT_SECURE_CONNECTION,
+                       ?CLIENT_SECURE_CONNECTION bor
+                       ?CLIENT_MULTI_STATEMENTS bor
+                       ?CLIENT_MULTI_RESULTS bor
+                       ?CLIENT_PS_MULTI_RESULTS,
     CapabilityFlags = case Database of
         undefined -> CapabilityFlags0;
         _         -> CapabilityFlags0 bor ?CLIENT_CONNECT_WITH_DB
@@ -328,6 +292,42 @@ parse_handshake_confirm(Packet) ->
 
 %% -- both text and binary protocol --
 
+%% @doc Fetches one or more results and and parses the result set(s) using
+%% either the text format (for plain queries) or the binary format (for
+%% prepared statements).
+-spec fetch_response(atom(), term(), timeout(), text | binary, list()) ->
+    {ok, [#ok{} | #resultset{} | #error{}]} | {error, timeout}.
+fetch_response(TcpModule, Socket, Timeout, Proto, Acc) ->
+    case recv_packet(TcpModule, Socket, Timeout, any) of
+        {ok, Packet, SeqNum2} ->
+            Result = case Packet of
+                ?ok_pattern ->
+                    parse_ok_packet(Packet);
+                ?error_pattern ->
+                    parse_error_packet(Packet);
+                ResultPacket ->
+                    %% The first packet in a resultset is only the column count.
+                    {ColCount, <<>>} = lenenc_int(ResultPacket),
+                    R0 = fetch_resultset(TcpModule, Socket, ColCount, SeqNum2),
+                    case R0 of
+                        #error{} = E ->
+                            %% TODO: Find a way to get here + testcase
+                            E;
+                        #resultset{} = R ->
+                            parse_resultset(R, ColCount, Proto)
+                    end
+            end,
+            Acc1 = [Result | Acc],
+            case more_results_exists(Result) of
+                true ->
+                    fetch_response(TcpModule, Socket, Timeout, Proto, Acc1);
+                false ->
+                    {ok, lists:reverse(Acc1)}
+            end;
+        {error, timeout} ->
+            {error, timeout}
+    end.
+
 %% @doc Fetches packets for a result set. The column definitions are parsed but
 %% the rows are unparsed binary packages. This function is used for both the
 %% text protocol and the binary protocol. This affects the way the rows need to
@@ -348,6 +348,22 @@ fetch_resultset(TcpModule, Socket, FieldCount, SeqNum) ->
             E
     end.
 
+parse_resultset(#resultset{cols = ColDefs, rows = Rows} = R, ColumnCount, text) ->
+    %% Parse the rows according to the 'text protocol' representation.
+    Rows1 = [decode_text_row(ColumnCount, ColDefs, Row) || Row <- Rows],
+    R#resultset{rows = Rows1};
+parse_resultset(#resultset{cols = ColDefs, rows = Rows} = R, ColumnCount, binary) ->
+    %% Parse the rows according to the 'binary protocol' representation.
+    Rows1 = [decode_binary_row(ColumnCount, ColDefs, Row) || Row <- Rows],
+    R#resultset{rows = Rows1}.
+
+more_results_exists(#ok{status = S}) ->
+    S band ?SERVER_MORE_RESULTS_EXISTS /= 0;
+more_results_exists(#error{}) ->
+    false; %% No status bits for error
+more_results_exists(#resultset{status = S}) ->
+    S band ?SERVER_MORE_RESULTS_EXISTS /= 0.
+
 %% @doc Receives NumLeft column definition packets. They are not parsed.
 %% @see parse_column_definition/1
 -spec fetch_column_definitions(atom(), term(), SeqNum :: integer(),
@@ -980,7 +996,6 @@ nulterm_str(Bin) ->
 
 -ifdef(TEST).
 -include_lib("eunit/include/eunit.hrl").
--include("server_status.hrl").
 
 %% Testing some of the internal functions, mostly the cases we don't cover in
 %% other tests.

test/mysql_protocol_tests.erl

@@ -43,7 +43,7 @@ resultset_test() ->
     ExpectedCommunication = [{send, ExpectedReq},
                              {recv, ExpectedResponse}],
     Sock = mock_tcp:create(ExpectedCommunication),
-    ResultSet = mysql_protocol:query(Query, mock_tcp, Sock, infinity),
+    {ok, [ResultSet]} = mysql_protocol:query(Query, mock_tcp, Sock, infinity),
     mock_tcp:close(Sock),
     ?assertMatch(#resultset{cols = [#col{name = <<"@@version_comment">>}],
                             rows = [[<<"MySQL Community Server (GPL)">>]]},
@@ -80,7 +80,7 @@ resultset_error_test() ->
         "48 04 23 48 59 30 30 30    4e 6f 20 74 61 62 6c 65    H.#HY000No table"
         "73 20 75 73 65 64                                     s used"),
     Sock = mock_tcp:create([{send, ExpectedReq}, {recv, ExpectedResponse}]),
-    Result = mysql_protocol:query(Query, mock_tcp, Sock, infinity),
+    {ok, [Result]} = mysql_protocol:query(Query, mock_tcp, Sock, infinity),
     ?assertMatch(#error{}, Result),
     mock_tcp:close(Sock),
     ok.

test/mysql_tests.erl

@@ -116,6 +116,7 @@ query_test_() ->
           {"Autocommit",           fun () -> autocommit(Pid) end},
           {"Encode",               fun () -> encode(Pid) end},
           {"Basic queries",        fun () -> basic_queries(Pid) end},
+          {"Multi statements",     fun () -> multi_statements(Pid) end},
           {"Text protocol",        fun () -> text_protocol(Pid) end},
           {"Binary protocol",      fun () -> binary_protocol(Pid) end},
           {"FLOAT rounding",       fun () -> float_rounding(Pid) end},
@@ -154,7 +155,7 @@ log_warnings_test() ->
     ?assertEqual("Warning 1364: Field 'x' doesn't have a default value\n"
                  " in INSeRT INtO foo () VaLUeS ()\n", Log2),
     ?assertEqual("Warning 1364: Field 'x' doesn't have a default value\n"
-                 " in prepared statement insrt\n", Log3),
+                 " in INSERT INTO foo () VALUES ()\n", Log3),
     exit(Pid, normal).
 
 autocommit(Pid) ->
@@ -191,6 +192,42 @@ basic_queries(Pid) ->
 
     ok.
 
+multi_statements(Pid) ->
+    %% Multiple statements, no result set
+    ?assertEqual(ok, mysql:query(Pid, "CREATE TABLE foo (bar INT);"
+                                      "DROP TABLE foo;")),
+
+    %% Multiple statements, one result set
+    ?assertEqual({ok, [<<"foo">>], [[42]]},
+                 mysql:query(Pid, "CREATE TABLE foo (bar INT);"
+                                  "DROP TABLE foo;"
+                                  "SELECT 42 AS foo;")),
+
+    %% Multiple statements, multiple result sets
+    ?assertEqual({ok, [{[<<"foo">>], [[42]]}, {[<<"bar">>], [[<<"baz">>]]}]},
+                 mysql:query(Pid, "SELECT 42 AS foo; SELECT 'baz' AS bar;")),
+
+    %% Multiple results in a prepared statement.
+    %% Preparing "SELECT ...; SELECT ...;" gives a syntax error although the
+    %% docs say it should be possible.
+
+    %% Instead, test executing a stored procedure that returns multiple result
+    %% sets using a prepared statement.
+
+    CreateProc = "CREATE PROCEDURE multifoo() BEGIN\n"
+                 "  SELECT 42 AS foo;\n"
+                 "  SELECT 'baz' AS bar;\n"
+                 "END;\n",
+    ok = mysql:query(Pid, CreateProc),
+    ?assertEqual({ok, multifoo},
+                 mysql:prepare(Pid, multifoo, "CALL multifoo();")),
+    ?assertEqual({ok, [{[<<"foo">>], [[42]]}, {[<<"bar">>], [[<<"baz">>]]}]},
+                 mysql:execute(Pid, multifoo, [])),
+    ?assertEqual(ok, mysql:unprepare(Pid, multifoo)),
+    ?assertEqual(ok, mysql:query(Pid, "DROP PROCEDURE multifoo;")),
+
+    ok.
+
 text_protocol(Pid) ->
     ok = mysql:query(Pid, ?create_table_t),
     ok = mysql:query(Pid, <<"INSERT INTO t (bl, f, d, dc, y, ti, ts, da, c)"