Initial Tables.jl implementation (#117)

* Initial Tables.jl implementation

* Remove DataStreams, add Tables to REQUIRE, fix tests

* Try to fix CI

* Remove osx testing for now

.appveyor.yml

@@ -7,6 +7,9 @@ platform:
   - x86 # 32-bit
   - x64 # 64-bit
 
+services:
+  - mysql
+
 # # Uncomment the following lines to allow failures on nightly julia
 # # (tests will run but not make your overall status red)
 # matrix:

.travis.yml

@@ -6,7 +6,6 @@ services:
 
 os:
   - linux
-  - osx
 
 julia:
   - 1.0
@@ -24,8 +23,6 @@ after_success:
 before_script:
 - export OLD_PATH=$LD_LIBRARY_PATH
 - export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:`mysql_config --libs | cut -d ' ' -f1 | sed 's/-L//'`
-- if [[ "$TRAVIS_OS_NAME" == "osx" ]]; then brew install mysql          ; fi
-- if [[ "$TRAVIS_OS_NAME" == "osx" ]]; then mysql.server start          ; fi
 
 after_script:
 - export LD_LIBRARY_PATH=$OLD_PATH

REQUIRE

@@ -1,5 +1,4 @@
 julia 0.7
-DataStreams
-Missings
+Tables
 DecFP
 BinaryProvider

src/MySQL.jl

@@ -1,8 +1,6 @@
 module MySQL
 
-using DataStreams, Missings, Dates
-
-export Data
+using Dates, Tables
 
 abstract type MySQLError end
 # For errors that happen in MySQL.jl
@@ -67,8 +65,8 @@ end
 Escapes a string using `mysql_real_escape_string()`, returns the escaped string.
 """
 function escape(conn::MySQL.Connection, str::String)
-    output = Vector{UInt8}(undef, length(str) * 2 + 1)
-    output_len = API.mysql_real_escape_string(conn.ptr, output, str, Culong(length(str)))
+    output = Vector{UInt8}(undef, sizeof(str) * 2 + 1)
+    output_len = API.mysql_real_escape_string(conn.ptr, output, str, Culong(sizeof(str)))
     if output_len == typemax(Cuint)
         throw(MySQLInternalError(conn))
     end
@@ -105,20 +103,41 @@ See list of DataStreams implementations [here](https://github.com/JuliaData/Data
 """
 function query end
 
-function query(conn::Connection, sql::String, sink::Type=Data.Table, args...; append::Bool=false, kwargs...)
-    source = Query(conn, sql; kwargs...)
-    sink = Data.stream!(source, sink, args...; append=append)
-    return Data.close!(sink)
+function query(conn::Connection, sql::String, sink::Union{Type, Nothing}=nothing, args...; append::Bool=false, kwargs...)
+    if sink === nothing
+        Base.depwarn("`MySQL.query(conn, sql)` will return a MySQL.Query in the future; to materialize the result, use `MySQL.query(conn, sql) |> columntable` or `MySQL.query(conn, sql) |> DataFrame` instead", nothing)
+        sink = columntable
+    else
+        Base.depwarn("`MySQL.query(conn, sql, $sink)` is deprecated; use `MySQL.query(conn, sql) |> $sink(args...)` instead", nothing)
+    end
+    if append
+        Base.depwarn("`append=true` is deprecated; use sink-specific append features instead. For example, `columntable(existing, MySQL.query(conn, sql))` or `append!(existing_df, MySQL.query(conn, sql))`")
+    end
+    return Query(conn, sql; kwargs...) |> sink
 end
 
 function query(conn::Connection, sql::String, sink::T; append::Bool=false, kwargs...) where {T}
-    source = Query(conn, sql; kwargs...)
-    sink = Data.stream!(source, sink; append=append)
-    return Data.close!(sink)
+    Base.depwarn("`MySQL.query(conn, sql, ::$T)` is deprecated; `MySQL.Query` now supports the Tables.jl interface, so any valid Tables.jl sink can receive a resultset", nothing)
+    if append
+        Base.depwarn("`append=true` is deprecated; use sink-specific append features instead. For example, `columntable(existing, MySQL.query(conn, sql))` or `append!(existing_df, MySQL.query(conn, sql))`")
+    end
+    return Query(conn, sql; kwargs...) |> T
 end
 
-query(source::Query, sink=Data.Table, args...; append::Bool=false, transforms::Dict=Dict{Int,Function}()) = (sink = Data.stream!(source, sink, args...; append=append, transforms=transforms); return Data.close!(sink))
-query(source::Query, sink::T; append::Bool=false, transforms::Dict=Dict{Int,Function}()) where {T} = (sink = Data.stream!(source, sink; append=append, transforms=transforms); return Data.close!(sink))
+function query(source::Query, sink=columntable, args...; append::Bool=false)
+    Base.depwarn("`MySQL.query(q::MySQL.Query)` is deprecated and will be removed in the future; `MySQL.Query` itself will iterate rows as NamedTuples and supports the Tables.jl interface", nothing)
+    if append
+        Base.depwarn("`append=true` is deprecated; use sink-specific append features instead. For example, `columntable(existing, MySQL.query(conn, sql))` or `append!(existing_df, MySQL.query(conn, sql))`")
+    end
+    return source |> sink
+end
+function query(source::Query, sink::T; append::Bool=false) where {T}
+    Base.depwarn("`MySQL.query(q::MySQL.Query)` is deprecated and will be removed in the future; `MySQL.Query` itself will iterate rows as NamedTuples and supports the Tables.jl interface", nothing)
+    if append
+        Base.depwarn("`append=true` is deprecated; use sink-specific append features instead. For example, `columntable(existing, MySQL.query(conn, sql))` or `append!(existing_df, MySQL.query(conn, sql))`")
+    end
+    return source |> T
+end
 
 include("prepared.jl")
 

src/api.jl

@@ -1,6 +1,6 @@
 module API
 
-using Dates, DecFP, Missings
+using Dates, DecFP
 
 # Load libmariadb from our deps.jl
 const depsjl_path = joinpath(dirname(@__FILE__), "..", "deps", "deps.jl")

src/prepared.jl

@@ -48,15 +48,26 @@ function execute!(stmt::Stmt, params=[])
     return API.mysql_stmt_affected_rows(stmt.ptr)
 end
 
-Data.streamtypes(::Type{Stmt}) = [Data.Row]
-function Data.streamto!(sink::Stmt, ::Type{Data.Row}, val, row, col)
-    sink.rows_affected += execute!(sink, val)
-    return
-end
-
-Stmt(sch::Data.Schema, ::Type{Data.Row}, append::Bool, conn::Connection, sql::String) = Stmt(conn, sql)
-Stmt(sink::Stmt, sch::Data.Schema, ::Type{Data.Row}, append::Bool) = sink
+execute!(itr, conn::Connection, sql::String) = execute!(itr, Stmt(conn, sql))
+function execute!(itr, stmt::Stmt)
+    rows = Tables.rows(itr)
+    state = iterate(rows)
+    state === nothing && return stmt
+    row, st = state
+    sch = Tables.Schema(propertynames(row), nothing)
+    binds = Vector{API.MYSQL_BIND}(undef, stmt.nparams)
+    bindptr = pointer(binds)
 
-function MySQLStatementIterator(args...)
-    throw(ArgumentError("`MySQLStatementIterator` is deprecated; instead, you can create a prepared statement by doing `stmt = MySQL.Stmt(conn, sql)` and then \"stream\" parameters to it, with the statement being executed once for each row in the source, like `Data.stream!(source, stmt)`"))
-end
+    while true
+        Tables.eachcolumn(sch, row) do val, col, nm
+            binds[col] = bind(val)
+        end
+        API.mysql_stmt_bind_param(stmt.ptr, bindptr) == 0 || throw(MySQLStatementError(stmt.ptr))
+        API.mysql_stmt_execute(stmt.ptr) == 0 || throw(MySQLStatementError(stmt.ptr))
+        stmt.rows_affected += API.mysql_stmt_affected_rows(stmt.ptr)
+        state = iterate(rows, st)
+        state === nothing && break
+        row, st = state
+    end
+    return stmt
+end