Use new BufferedIO type in Parsers for IO inputs for performance. Add… (#355)

* Use new BufferedIO type in Parsers for IO inputs for performance. Addresses #350

* Require Parsers 0.2.13

* Bump Parsers dependency

Author: quinnj

REQUIRE

@@ -4,4 +4,4 @@ DataFrames 0.14.0
 WeakRefStrings 0.4.1
 CategoricalArrays 0.3.0
 Tables
-Parsers
+Parsers 0.2.14

src/CSV.jl

@@ -190,16 +190,16 @@ function File(source::Union{String, IO};
         end
     end
 
-    !transpose && seek(io, positions[1])
+    !transpose && Parsers.fastseek!(io, positions[1])
     return File{transpose, columnaccess, typeof(io), typeof(parsinglayers), typeof(kwargs)}(names, finaltypes, getname(source), io, parsinglayers, positions, originalpositions, Ref(1), ref, Ref(Parsers.SUCCESS), kwargs, pools, strict, silencewarnings)
 end
 
 include("filedetection.jl")
 include("typedetection.jl")
 include("tables.jl")
 
-getio(str::String, use_mmap) = IOBuffer(use_mmap ? Mmap.mmap(str) : Base.read(str))
-getio(io::IO, use_mmap) = io
+getio(str::String, use_mmap) = use_mmap ? IOBuffer(Mmap.mmap(str)) : Parsers.BufferedIO(open(str))
+getio(io::IO, use_mmap) = Parsers.BufferedIO(io)
 getname(str::String) = str
 getname(io::I) where {I <: IO} = string("<", I, ">")
 
@@ -208,8 +208,8 @@ function consumeBOM!(io)
     startpos = position(io)
     if !eof(io) && Parsers.peekbyte(io) == 0xef
         Parsers.readbyte(io)
-        (!eof(io) && Parsers.readbyte(io) == 0xbb) || seek(io, startpos)
-        (!eof(io) && Parsers.readbyte(io) == 0xbf) || seek(io, startpos)
+        (!eof(io) && Parsers.readbyte(io) == 0xbb) || Parsers.fastseek!(io, startpos)
+        (!eof(io) && Parsers.readbyte(io) == 0xbf) || Parsers.fastseek!(io, startpos)
     end
     return
 end
@@ -310,7 +310,7 @@ function read(fullpath::Union{AbstractString,IO}, sink=DataFrame, args...; appen
     f = CSV.File(fullpath; kwargs...)
     if !isempty(transforms)
         Base.depwarn("`CSV.read(source; transforms=Dict(...)` is deprecated in favor of `CSV.File(source) |> transform(col1=x->...) |> DataFrame`", nothing)
-        return f |> transform(transforms) |> sink
+        return sink(transform(f, transforms))
     end
     return f |> sink
 end

src/filedetection.jl

@@ -76,7 +76,7 @@ function datalayout_transpose(header, parsinglayers, io, datarow, footerskip, no
             push!(columnpositions, position(io))
             readline!(parsinglayers, io)
         end
-        seek(io, datapos)
+        Parsers.fastseek!(io, datapos)
     elseif isa(header, AbstractRange)
         # column names span several columns
         throw(ArgumentError("not implemented for transposed csv files"))
@@ -104,7 +104,7 @@ function datalayout_transpose(header, parsinglayers, io, datarow, footerskip, no
             push!(columnpositions, position(io))
             readline!(parsinglayers, io)
         end
-        seek(io, datapos)
+        Parsers.fastseek!(io, datapos)
     end
     rows = rows - footerskip # rows now equals the actual number of rows in the dataset
     return rows, makeunique(map(x->normalizenames ? normalizename(x) : Symbol(x), columnnames)), columnpositions
@@ -117,7 +117,7 @@ function datalayout(header::Integer, parsinglayers, io, datarow, normalizenames,
         skipto!(parsinglayers, io, 1, datarow)
         datapos = position(io)
         row_vals = readsplitline(parsinglayers, io, cmt)
-        seek(io, datapos)
+        Parsers.fastseek!(io, datapos)
         columnnames = [Symbol("Column$i") for i = eachindex(row_vals)]
     else
         skipto!(parsinglayers, io, 1, header)
@@ -148,7 +148,7 @@ function datalayout(header::Vector, parsinglayers, io, datarow, normalizenames,
         columnnames = makeunique([normalizenames ? normalizename(nm) : Symbol(nm) for nm in header])
     else
         row_vals = readsplitline(parsinglayers, io, cmt)
-        seek(io, datapos)
+        Parsers.fastseek!(io, datapos)
         if isempty(header)
             columnnames = [Symbol("Column$i") for i in eachindex(row_vals)]
         else

src/transforms.jl

@@ -53,7 +53,7 @@ Base.IteratorSize(::Type{<:Transforms{T}}) where {T} = Base.IteratorSize(T)
 Base.length(t::Transforms) = length(getfield(t, 1))
 Base.eltype(t::Transforms{T, F}) where {T, F} = TransformsRow{eltype(getfield(t, 1)), F}
 
-function Base.iterate(t::Transforms, st=())
+@inline function Base.iterate(t::Transforms, st=())
     state = iterate(getfield(t, 1), st...)
     state === nothing && return nothing
     return TransformsRow(state[1], getfield(t, 2)), (state[2],)

src/typedetection.jl

@@ -90,18 +90,18 @@ function detect(types, io, positions, parsinglayers, kwargs, typemap, categorica
     for startingrow in rng
         for row = trunc(Int64, startingrow):trunc(Int64, startingrow + step - 1)
             rows += 1
-            !transpose && seek(io, positions[row])
+            !transpose && Parsers.fastseek!(io, positions[row])
             lastcode[] = Parsers.SUCCESS
             for col = 1:cols
                 if !transpose && newline(lastcode[])
                     typecodes[col] = promote_typecode(typecodes[col], MISSING)
                     continue
                 end
-                transpose && seek(io, positions[col])
+                transpose && Parsers.fastseek!(io, positions[col])
                 # if debug
                 #     pos = position(io)
                 #     result = Parsers.parse(parsinglayers, io, String)
-                #     seek(io, pos)
+                #     Parsers.fastseek!(io, pos)
                 # end
                 @inbounds T = typecodes[col]
                 if T === USER
@@ -145,7 +145,7 @@ function incr!(dict::Dict{String, Int}, key::Tuple{Ptr{UInt8}, Int})
 end
 
 @inline function trytype(io, pos, layers, T, kwargs, lastcode)
-    seek(io, pos)
+    Parsers.fastseek!(io, pos)
     res = Parsers.parse(layers, io, T; kwargs...)
     lastcode[] = res.code
     return Parsers.ok(res.code) ? typecode(res.result) : EMPTY