Support direct DBInterface parameter binding

src/execute.jl

@@ -146,10 +146,11 @@ Close a cursor. No more results will be available.
 DBInterface.close!(c::TextCursor) = clear!(c.conn)
 
 """
-    DBInterface.execute(conn::MySQL.Connection, sql) => MySQL.TextCursor
+    DBInterface.execute(conn::MySQL.Connection, sql, [params]) => DBInterface.Cursor
 
-Execute the SQL `sql` statement with the database connection `conn`. Parameter binding is
-only supported via prepared statements, see `?DBInterface.prepare(conn, sql)`.
+Execute the SQL `sql` statement with the database connection `conn`, optionally passing
+`params` to bind to parameter markers. Queries with parameters are prepared for this
+execution. Use `DBInterface.prepare` directly to reuse a statement across executions.
 Returns a `Cursor` object, which iterates resultset rows and satisfies the `Tables.jl` interface, meaning
 results can be sent to any valid sink function (`DataFrame(cursor)`, `CSV.write("results.csv", cursor)`, etc.).
 Specifying `mysql_store_result=false` will avoid buffering the full resultset to the client after executing
@@ -158,7 +159,7 @@ fetched one at a time.
 """
 function DBInterface.execute(conn::Connection, sql::AbstractString, params=(); mysql_store_result::Bool=true, mysql_date_and_time::Bool=false)
     checkconn(conn)
-    params != () && error("`DBInterface.execute(conn, sql)` does not support parameter binding; see `?DBInterface.prepare(conn, sql)`")
+    params != () && return executeparams(conn, sql, params; mysql_store_result, mysql_date_and_time)
     clear!(conn)
     API.query(conn.mysql, sql)
 

src/prepare.jl

@@ -72,6 +72,7 @@ function DBInterface.prepare(conn::Connection, sql::AbstractString; mysql_date_a
 end
 
 mutable struct Cursor{buffered} <: DBInterface.Cursor
+    conn::Connection
     stmt::API.MYSQL_STMT
     nfields::Int
     names::Vector{Symbol}
@@ -82,6 +83,7 @@ mutable struct Cursor{buffered} <: DBInterface.Cursor
     rows_affected::Int64
     rows::Int
     current_rownumber::Int
+    statement::Union{Nothing, Statement}
 end
 
 struct Row <: Tables.AbstractRow
@@ -135,7 +137,16 @@ end
 
 Close a cursor. No more results will be available.
 """
-DBInterface.close!(c::Cursor) = clear!(c.conn)
+function DBInterface.close!(c::Cursor)
+    if c.statement === nothing
+        c.conn.mysql.ptr == C_NULL || clear!(c.conn)
+    elseif c.stmt.ptr != C_NULL
+        c.conn.mysql.ptr == C_NULL || clear!(c.conn, c.stmt)
+        API.close!(c.stmt)
+        c.conn.lastexecute === c.stmt && (c.conn.lastexecute = nothing)
+    end
+    return
+end
 
 @noinline paramcheck(stmt, args) = length(args) == stmt.nparams || throw(MySQLInterfaceError("stmt requires $(stmt.nparams) params, only $(length(args)) provided"))
 
@@ -191,7 +202,19 @@ function DBInterface.execute(stmt::Statement, params=(); mysql_store_result::Boo
             lookup = Dict(x => i for (i, x) in enumerate(names))
         end
     end
-    return Cursor{buffered}(stmt.stmt, nfields, names, types, lookup, valuehelpers, values, rows_affected, rows, 0)
+    return Cursor{buffered}(stmt.conn, stmt.stmt, nfields, names, types, lookup, valuehelpers, values, rows_affected, rows, 0, nothing)
+end
+
+function executeparams(conn::Connection, sql::AbstractString, params; mysql_store_result::Bool, mysql_date_and_time::Bool)
+    stmt = DBInterface.prepare(conn, sql; mysql_date_and_time)
+    try
+        cursor = DBInterface.execute(stmt, params; mysql_store_result, mysql_date_and_time)
+        cursor.statement = stmt
+        return cursor
+    catch
+        DBInterface.close!(stmt)
+        rethrow()
+    end
 end
 
 inithelper!(helper, x::Missing) = nothing

test/runtests.jl

@@ -344,6 +344,26 @@ DBInterface.close!(stmt)
 res = DBInterface.execute(conn, "SELECT value FROM NullBindingTest") |> columntable
 @test isequal(res.value, [missing, 1, missing])
 
+@testset "connection-level parameter binding (#238)" begin
+    DBInterface.execute(conn, "CREATE TABLE DirectExecuteTest (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(32), value INT NULL)")
+    cursor = DBInterface.execute(conn, "INSERT INTO DirectExecuteTest (name, value) VALUES (?, ?)", ("first", nothing))
+    @test cursor isa MySQL.Cursor
+    @test DBInterface.lastrowid(cursor) == 1
+    @test DBInterface.close!(cursor) === nothing
+
+    DBInterface.execute(conn, "INSERT INTO DirectExecuteTest (name, value) VALUES (?, ?)", ("second", 2))
+    cursor = DBInterface.execute(conn, "SELECT name, value FROM DirectExecuteTest WHERE id >= ? ORDER BY id", (1,))
+    GC.gc()
+    result = Tables.columntable(cursor)
+    @test result.name == ["first", "second"]
+    @test isequal(result.value, [missing, 2])
+
+    cursor = DBInterface.execute(conn, "SELECT name FROM DirectExecuteTest WHERE id >= ? ORDER BY id", (1,); mysql_store_result=false)
+    @test first(cursor).name == "first"
+    @test DBInterface.close!(cursor) === nothing
+    @test Tables.columntable(DBInterface.execute(conn, "SELECT COUNT(*) AS count FROM DirectExecuteTest")).count == [2]
+end
+
 stmt = DBInterface.prepare(conn, "select * from Employee")
 res = DBInterface.execute(stmt) |> columntable
 DBInterface.close!(stmt)
@@ -421,6 +441,8 @@ res = DBInterface.execute(resstmt) |> columntable
 res = DBInterface.execute(conn, "select id, t from datetime6_field"; mysql_date_and_time=true) |> columntable
 @test length(res) == 2
 @test res[2][1] == DateAndTime(Date(2021, 1, 2), Time(1, 2, 3, 456, 789))
+res = DBInterface.execute(conn, "select id, t from datetime6_field where id = ?", (1,); mysql_date_and_time=true) |> columntable
+@test res[2][1] == DateAndTime(Date(2021, 1, 2), Time(1, 2, 3, 456, 789))
 
 DBInterface.execute(conn, """
 CREATE PROCEDURE get_employee()