Implement new DBInterface.executemultiple interface function (#155)

This allows returning multiple resultsets from a single query call; this
requires setting `multi_statements` to `true` by default, but I couldn't
think of any downside to doing this (vs. the alternative in which a user
finds the function, tries it and immediately fails and has to restart
their connection by setting the flag). Prepared statements don't support
multiple resultsets, and an error is thrown appropriately (from mysql
itself). This PR requires
JuliaDatabases/DBInterface.jl#26 to be merged
and tagged first.

Author: quinnj

src/MySQL.jl

@@ -47,7 +47,16 @@ end
 
 function clear!(conn, result::API.MYSQL_RES)
     if result.ptr != C_NULL
-        while API.fetchrow(conn.mysql, result) != C_NULL || API.nextresult(conn.mysql) !== nothing
+        while true
+            if API.fetchrow(conn.mysql, result) == C_NULL
+                if API.moreresults(conn.mysql)
+                    finalize(result)
+                    @assert API.nextresult(conn.mysql) !== nothing
+                    result = API.useresult(conn.mysql)
+                else
+                    break
+                end
+            end
         end
         finalize(result)
     end
@@ -68,7 +77,7 @@ function clientflags(;
         compress::Bool=false,
         ignore_space::Bool=false,
         local_files::Bool=false,
-        multi_statements::Bool=false,
+        multi_statements::Bool=true,
         multi_results::Bool=false,
         kw...
     )

src/execute.jl

@@ -9,19 +9,22 @@ mutable struct TextCursor{buffered} <: DBInterface.Cursor
     types::Vector{Type}
     lookup::Dict{Symbol, Int}
     current_rownumber::Int
+    current_resultsetnumber::Int
 end
 
 struct TextRow{buffered} <: Tables.AbstractRow
     cursor::TextCursor{buffered}
     row::Ptr{Ptr{UInt8}}
     lengths::Vector{Culong}
     rownumber::Int
+    resultsetnumber::Int
 end
 
 getcursor(r::TextRow) = getfield(r, :cursor)
 getrow(r::TextRow) = getfield(r, :row)
 getlengths(r::TextRow) = getfield(r, :lengths)
 getrownumber(r::TextRow) = getfield(r, :rownumber)
+getresultsetnumber(r::TextRow) = getfield(r, :resultsetnumber)
 
 Tables.columnnames(r::TextRow) = getcursor(r).names
 
@@ -71,7 +74,7 @@ end
 @noinline wrongrow(i) = throw(ArgumentError("row $i is no longer valid; mysql results are forward-only iterators where each row is only valid when iterated"))
 
 function Tables.getcolumn(r::TextRow, ::Type{T}, i::Int, nm::Symbol) where {T}
-    getrownumber(r) == getcursor(r).current_rownumber || wrongrow(getrownumber(r))
+    (getrownumber(r) == getcursor(r).current_rownumber && getresultsetnumber(r) == getcursor(r).current_resultsetnumber) || wrongrow(getrownumber(r))
     return cast(T, unsafe_load(getrow(r), i), getlengths(r)[i])
 end
 
@@ -91,18 +94,11 @@ function Base.iterate(cursor::TextCursor{buffered}, i=1) where {buffered}
     rowptr = API.fetchrow(cursor.conn.mysql, cursor.result)
     if rowptr == C_NULL
         !buffered && API.errno(cursor.conn.mysql) != 0 && throw(API.Error(cursor.conn.mysql))
-        if API.nextresult(cursor.conn.mysql) === nothing
-            finalize(cursor.result)
-            return nothing
-        else
-            # we ***ignore*** additional resultsets for now
-            clear!(cursor.conn)
-            return nothing
-        end
+        return nothing
     end
     lengths = API.fetchlengths(cursor.result, cursor.nfields)
     cursor.current_rownumber = i
-    return TextRow(cursor, rowptr, lengths, i), i + 1
+    return TextRow(cursor, rowptr, lengths, i, cursor.current_resultsetnumber), i + 1
 end
 
 """
@@ -167,5 +163,36 @@ function DBInterface.execute(conn::Connection, sql::AbstractString, params=(); m
         error("error with mysql resultset columns")
     end
     lookup = Dict(x => i for (i, x) in enumerate(names))
-    return TextCursor{buffered}(conn, sql, nfields, nrows, Core.bitcast(Int64, rows_affected), result, names, types, lookup, 0)
-end
+    return TextCursor{buffered}(conn, sql, nfields, nrows, Core.bitcast(Int64, rows_affected), result, names, types, lookup, 0, 1)
+end
+
+struct TextCursors{T}
+    cursor::TextCursor{T}
+end
+
+Base.eltype(c::TextCursors{T}) where {T} = TextCursor{T}
+Base.IteratorSize(::Type{<:TextCursors}) = Base.SizeUnknown()
+
+function Base.iterate(cursor::TextCursors{buffered}, first=true) where {buffered}
+    cursor.cursor.result.ptr == C_NULL && return nothing
+    if !first
+        finalize(cursor.cursor.result)
+        if API.moreresults(cursor.cursor.conn.mysql)
+            @assert API.nextresult(cursor.cursor.conn.mysql) !== nothing
+            cursor.cursor.result = buffered ? API.storeresult(cursor.cursor.conn.mysql) : API.useresult(cursor.cursor.conn.mysql)
+            if buffered
+                cursor.cursor.nrows = API.numrows(cursor.cursor.result)
+            end
+            cursor.cursor.nfields = API.numfields(cursor.cursor.result)
+            fields = API.fetchfields(cursor.cursor.result, cursor.cursor.nfields)
+            cursor.cursor.names = [ccall(:jl_symbol_n, Ref{Symbol}, (Ptr{UInt8}, Csize_t), x.name, x.name_length) for x in fields]
+            cursor.cursor.types = [juliatype(x.field_type, API.notnullable(x), API.isunsigned(x), API.isbinary(x)) for x in fields]
+        else
+            return nothing
+        end
+    end
+    return cursor.cursor, false
+end
+
+DBInterface.executemultiple(conn::Connection, sql::AbstractString, params=(); kw...) =
+    TextCursors(DBInterface.execute(conn, sql, params; kw...))

test/runtests.jl

@@ -249,3 +249,22 @@ res = DBInterface.execute(stmt) |> columntable
 @test length(res[1]) == 5
 res = DBInterface.execute(stmt)
 res = DBInterface.execute(stmt)
+
+results = DBInterface.executemultiple(conn, "select * from Employee; select DeptNo, OfficeNo from Employee where OfficeNo IS NOT NULL")
+state = iterate(results)
+@test state !== nothing
+res, st = state
+@test !st
+@test length(res) == 5
+ret = columntable(res)
+@test length(ret[1]) == 5
+state = iterate(results, st)
+@test state !== nothing
+res, st = state
+@test !st
+@test length(res) == 4
+ret = columntable(res)
+@test length(ret[1]) == 4
+
+# multiple-queries not supported by mysql w/ prepared statements
+@test_throws MySQL.API.StmtError DBInterface.prepare(conn, "select * from Employee; select DeptNo, OfficeNo from Employee where OfficeNo IS NOT NULL")