Streaming queries (#124)

For large datasets, often don't want to load the full set at once, this allows fetching and processing rows one at a time

Author: mcmcgrath13

.travis.yml

@@ -8,7 +8,7 @@ os:
   - linux
 
 julia:
-  - 1.0
+  - 1.1
   - nightly
 
 notifications:

Project.toml

@@ -7,6 +7,7 @@ version = "0.7.0"
 BinaryProvider = "b99e7846-7c00-51b0-8f62-c81ae34c0232"
 Dates = "ade2ca70-3891-5945-98fb-dc099432e06a"
 DecFP = "55939f99-70c6-5e9b-8bb0-5071ed7d61fd"
+Libdl = "8f399da3-3557-5675-b5ff-fb832c97cbdb"
 Tables = "bd369af6-aec1-5ad0-b16a-f7cc5008161c"
 
 [compat]

README.md

@@ -28,7 +28,7 @@ julia> Pkg.add("MySQL")
 
 ## Project Status
 
-The package is tested against the current Julia `1.0` release and nightly on Linux and OS X.
+The package is tested against the current Julia `1.1` release and nightly on Linux and OS X.
 
 ## Contributing and Questions
 
@@ -121,15 +121,9 @@ MySQL.escape(conn::MySQL.Connection, str::String) -> String
 ```
 Escape an SQL statement
 
-#### MySQL.Query (previously MySQL.query)
+#### MySQL.query (deprecated)
 
-```julia
-MySQL.Query(conn::MySQL.Connection, sql::String; append::Bool=false) => sink
-```
-Execute an SQL statement and return the results as a MySQL.Query object (see [MySQL.Query](#mysqlquery)).
-
-The results can be materialized as a data sink that implements the Tables.jl interface.
-E.g. `MySQL.Query(conn, sql) |> DataFrame` or `MySQL.Query(conn, sql) |> columntable`
+Deprecated - see [MySQL.Query](#mysqlquery)
 
 #### MySQL.execute!
 
@@ -175,10 +169,17 @@ Alternately, a source implementing the Tables.jl interface can be streamed by ex
 #### MySQL.Query
 
 ```julia
-MySQL.Query(conn, sql, sink=Data.Table, kwargs...) => MySQL.Query
+MySQL.Query(conn, sql, kwargs...) => MySQL.Query
 ```
 
-Execute an SQL statement and return a `MySQL.Query` object. Result rows can be iterated.
+Execute an SQL statement and return a `MySQL.Query` object. Result rows can be
+iterated as NamedTuples via `Table.rows(query)` where `query` is the `MySQL.Query`
+object.
+
+Supported Key Word Arguments:
+* `streaming` - Defaults to false. If true, length of the result size is unknown as the result is returned row by row. May be more memory efficient.
+
+To materialize the results as a `DataFrame`, use `MySQL.Query(conn, sql) |> DataFrame`.
 
 ### Example
 
@@ -189,7 +190,7 @@ using DataFrames
 
 conn = MySQL.connect("localhost", "root", "password", db = "test_db")
 
-foo = MySQL.query(conn, """SELECT COUNT(*) FROM my_first_table;""") |> DataFrame
+foo = MySQL.Query(conn, """SELECT COUNT(*) FROM my_first_table;""") |> DataFrame
 num_foo = foo[1,1]
 
 my_stmt = MySQL.Stmt(conn, """INSERT INTO my_second_table ('foo_id','foo_name') VALUES (?,?);""")

src/api.jl

@@ -19,7 +19,7 @@ const MYSQL_ROW = Ptr{Ptr{Cchar}}  # pointer to an array of strings
 const MYSQL_TYPE = UInt32
 
 """
-The field object that contains the metadata of the table. 
+The field object that contains the metadata of the table.
 Returned by mysql_fetch_fields API.
 """
 struct MYSQL_FIELD
@@ -114,8 +114,8 @@ struct MYSQL_BIND
     store_param_func::Ptr{Cvoid}
     fetch_result::Ptr{Cvoid}
     skip_result::Ptr{Cvoid}
-    buffer_length::Culong 
-    offset::Culong 
+    buffer_length::Culong
+    offset::Culong
     length_value::Culong
     param_number::Cuint
     pack_length::Cuint
@@ -191,20 +191,6 @@ macro c(func, ret, args, vals...)
     end
 end
 
-# function  mysql_library_init(argc=0, argv=C_NULL, groups=C_NULL)
-#     return ccall((:mysql_library_init, libmariadb),
-#                  Cint,
-#                  (Cint, Ptr{Ptr{UInt8}}, Ptr{Ptr{UInt8}}),
-#                  argc, argv, groups)
-# end
-
-# function  mysql_library_end()
-#     return ccall((:mysql_library_end, libmariadb),
-#                  Cvoid,
-#                  (),
-#                 )
-# end
-
 """
 Initializes the MYSQL object. Must be called before mysql_real_connect.
 Memory allocated by mysql_init can be freed with mysql_close.
@@ -428,6 +414,9 @@ function mysql_stmt_bind_result(stmtptr, bind::Ptr{MYSQL_BIND})
                  bind)
 end
 
+"""
+Submit a query to the server
+"""
 function mysql_query(mysqlptr::Ptr{Cvoid}, sql::String)
     return @c(:mysql_query,
                  Cint,
@@ -436,13 +425,26 @@ function mysql_query(mysqlptr::Ptr{Cvoid}, sql::String)
                  sql)
 end
 
+"""
+After mysql_query or mysql_real_query used to store result in memory and send all to client
+"""
 function mysql_store_result(mysqlptr::Ptr{Cvoid})
     return @c(:mysql_store_result,
                  MYSQL_RES,
                  (Ptr{Cvoid}, ),
                  mysqlptr)
 end
 
+"""
+After mysql_query or mysql_real_query used to stream result and send to client row by row
+"""
+function mysql_use_result(mysqlptr::Ptr{Cvoid})
+    return @c(:mysql_use_result,
+                 MYSQL_RES,
+                 (Ptr{Cvoid}, ),
+                 mysqlptr)
+end
+
 """
 Returns the field metadata.
 """

src/types.jl

@@ -48,7 +48,9 @@ function metadata(result::API.MYSQL_RES)
     return unsafe_wrap(Array, rawfields, nfields)
 end
 
-mutable struct Query{hasresult, names, T}
+# resulttype is a symbol with values :none :streaming :default
+# names and types relate to the returned columns in the query
+mutable struct Query{resulttype, names, T}
     result::Result
     ptr::Ptr{Ptr{Int8}}
     ncols::Int
@@ -62,39 +64,51 @@ function julia_type(field_type, notnullable, isunsigned)
 end
 
 """
-    MySQL.Query(conn, sql, sink=Data.Table; kwargs...) => MySQL.Query
+    MySQL.Query(conn, sql; kwargs...) => MySQL.Query
 
 Execute an SQL statement and return a `MySQL.Query` object. Result rows can be
-iterated as NamedTuples via `Data.rows(query)` where `query` is the `MySQL.Query`
+iterated as NamedTuples via `Table.rows(query)` where `query` is the `MySQL.Query`
 object.
 
-To materialize the results as a `DataFrame`, use `MySQL.query(conn, sql) |> DataFrame`.
+Supported Key Word Arguments:
+* `streaming` - Defaults to false. If true, length of the result size is unknown as the result is returned row by row. May be more memory efficient.
+
+To materialize the results as a `DataFrame`, use `MySQL.Query(conn, sql) |> DataFrame`.
 """
-function Query(conn::Connection, sql::String; kwargs...)
+function Query(conn::Connection, sql::String; streaming::Bool=false, kwargs...)
     conn.ptr == C_NULL && throw(MySQLInterfaceError("Method called with null connection."))
     MySQL.API.mysql_query(conn.ptr, sql) != 0 && throw(MySQLInternalError(conn))
-    result = MySQL.Result(MySQL.API.mysql_store_result(conn.ptr))
+
+    if streaming
+        resulttype = :streaming
+        result = MySQL.Result(MySQL.API.mysql_use_result(conn.ptr))
+    else
+        resulttype = :default
+        result = result = MySQL.Result(MySQL.API.mysql_store_result(conn.ptr))
+    end
+
     if result.ptr != C_NULL
         nrows = MySQL.API.mysql_num_rows(result.ptr)
         fields = MySQL.metadata(result.ptr)
         names = Tuple(ccall(:jl_symbol_n, Ref{Symbol}, (Ptr{UInt8}, Csize_t), x.name, x.name_length) for x in fields)
         T = Tuple{(julia_type(x.field_type, API.notnullable(x), API.isunsigned(x)) for x in fields)...}
-        hasresult = true
         ncols = length(fields)
         ptr = MySQL.API.mysql_fetch_row(result.ptr)
     elseif API.mysql_field_count(conn.ptr) == 0
         result = Result(Int(API.mysql_affected_rows(conn.ptr)))
         nrows = ncols = 1
         names = (:num_rows_affected,)
         T = Tuple{Int}
-        hasresult = false
+        resulttype = :none
         ptr = C_NULL
     else
         throw(MySQLInterfaceError("Query expected to produce results but did not."))
     end
-    return Query{hasresult, names, T}(result, ptr, ncols, nrows)
+    return Query{resulttype, names, T}(result, ptr, ncols, nrows)
 end
 
+Base.IteratorSize(::Type{Query{resulttype, names, T}}) where {resulttype, names, T} = resulttype == :streaming ? Base.SizeUnknown() : Base.HasLength()
+
 Tables.istable(::Type{<:Query}) = true
 Tables.rowaccess(::Type{<:Query}) = true
 Tables.rows(q::Query) = q
@@ -126,9 +140,8 @@ function generate_namedtuple(::Type{NamedTuple{names, types}}, q) where {names,
     end
 end
 
-function Base.iterate(q::Query{hasresult, names, types}, st=1) where {hasresult, names, types}
-    st > length(q) && return nothing
-    !hasresult && return (num_rows_affected=Int(q.result.ptr),), 2
+function Base.iterate(q::Query{resulttype, names, types}, st=1) where {resulttype, names, types}
+    st == 1 && resulttype == :none && return (num_rows_affected=Int(q.result.ptr),), 2
     q.ptr == C_NULL && return nothing
     nt = generate_namedtuple(NamedTuple{names, types}, q)
     q.ptr = API.mysql_fetch_row(q.result.ptr)

test/runtests.jl

@@ -1,12 +1,12 @@
 using Test, MySQL, Tables, Dates
 
 if haskey(ENV, "APPVEYOR_BUILD_NUMBER")
-    pwd = "Password12!"
+    pswd = "Password12!"
 else
-    pwd = ""
+    pswd = ""
 end
 
-const conn = MySQL.connect("127.0.0.1", "root", pwd; port=3306)
+const conn = MySQL.connect("127.0.0.1", "root", pswd; port=3306)
 
 MySQL.execute!(conn, "DROP DATABASE if exists mysqltest")
 MySQL.execute!(conn, "CREATE DATABASE mysqltest")
@@ -58,6 +58,19 @@ expected = (
 
 @test res == expected
 
+# Streaming Queries
+sres = MySQL.Query(conn, "select * from Employee", streaming=true)
+
+@test sres.nrows == 0
+
+data = []
+for row in sres
+    push!(data, row)
+end
+@test length(data) == 4
+@test length(data[1]) == 10
+@test data[1].Name == "John"
+
 # insert null row
 MySQL.execute!(conn, "INSERT INTO Employee () VALUES ();")