improve the performance of byrow(join)

Author: sl-solution

src/byrow/byrow.jl

@@ -143,7 +143,7 @@ byrow(ds::AbstractDataset, ::typeof(stdze!), cols::MultiColumnIndex = names(ds,
 byrow(ds::AbstractDataset, ::typeof(hash), cols::MultiColumnIndex = :; by = identity, threads = nrow(ds) > __NCORES*10) = row_hash(ds, by, cols, threads = threads)
 byrow(ds::AbstractDataset, ::typeof(hash), col::ColumnIndex; by = identity, threads = nrow(ds) > __NCORES*10) = byrow(ds, hash, [col]; by = by, threads = threads)
 
-byrow(ds::AbstractDataset, ::typeof(join), col::MultiColumnIndex; threads = nrow(ds) > __NCORES*10, delim = "", last = "") = row_join2(ds, col, threads = threads, delim = delim, last = last)
+byrow(ds::AbstractDataset, ::typeof(join), col::MultiColumnIndex; threads = nrow(ds) > __NCORES*10, delim = "", last = "") = row_join(ds, col, threads = threads, delim = delim, last = last)
 
 byrow(ds::AbstractDataset, ::typeof(mapreduce), cols::MultiColumnIndex = names(ds, Union{Missing, Number}); op = .+, f = identity,  init = missings(mapreduce(eltype, promote_type, view(_columns(ds),index(ds)[cols])), nrow(ds)), kwargs...) = mapreduce(f, op, eachcol(ds[!, cols]), init = init; kwargs...)
 

src/byrow/row_functions.jl

@@ -1112,40 +1112,52 @@ function row_hash(ds::AbstractDataset, f::Function, cols = :; threads = true)
 end
 row_hash(ds::AbstractDataset, cols = :; threads = true) = row_hash(ds, identity, cols; threads = threads)
 
-Base.@propagate_inbounds function _op_for_join!(x, y, delim, last, p, idx, lo, hi)
-    idx[] += 1
-    @simd for i in lo:hi
-        if idx[] == 1
-            x[i] = STRING(y[i])
-            x[i] *= delim
-        elseif idx[] < p
-            x[i] *= STRING(y[i])
-            x[i] *= delim
-        else
-            x[i] *= STRING(y[i])
-            x[i] *= last
+function _convert_uint8_to_string!(res, init0, curr_pos, ds, threads)
+    if threads
+        Threads.@threads for i in 1:nrow(ds)
+            res[i] = String(view(init0, 1:curr_pos[i]-1, i))
+        end
+    else
+        for i in 1:nrow(ds)
+            res[i] = String(view(init0, 1:curr_pos[i]-1, i))
+        end
+    end
+end
+function _add_last_for_join!(init0, curr_pos, ds, last_uint, last_len, threads)
+    if threads
+        Threads.@threads for i in 1:nrow(ds)
+            init0[curr_pos[i]-1:curr_pos[i]+last_len-2, i] = last_uint
+            curr_pos[i] += last_len-1
+        end
+    else
+        for i in 1:nrow(ds)
+            init0[curr_pos[i]-1:curr_pos[i]+last_len-2, i] = last_uint
+            curr_pos[i] += last_len-1
         end
     end
-    x
 end
 
-function row_join2(ds::AbstractDataset, cols = :; threads = true, delim = ",", last = "")
+function row_join(ds::AbstractDataset, cols = :; threads = true, delim::AbstractString = ",", last::AbstractString = "")
     colsidx = multiple_getindex(index(ds), cols)
-    init0 = Vector{Union{Missing, String}}(undef, nrow(ds))
 
-    if threads
-        cz = div(length(init0), __NCORES)
-        idx = [Ref{Int}(0) for _ in 1:__NCORES]
-        Threads.@threads for i in 1:__NCORES
-            lo = (i-1)*cz+1
-            i == __NCORES ? hi = length(init0) : hi = i*cz
-            mapreduce(identity, (x,y) -> _op_for_join!(x, y, delim, last, length(colsidx), idx[i], lo, hi), view(_columns(ds),colsidx), init = init0)
-        end
+    max_line_size = maximum(byrow(ds, sum, colsidx, by = y->length(__STRING(y)), threads = threads))
+    max_line_size += length(delim)*(length(colsidx)) + length(last)+1
+    init0 = Matrix{UInt8}(undef, max_line_size, nrow(ds))
+    curr_pos = ones(Int, nrow(ds))
+
+    delimiter = Base.CodeUnits(delim)
+    row_join!(init0, curr_pos, ds, repeat([identity], length(colsidx)), colsidx; delim = delimiter, quotechar = nothing, threads = threads)
+    if length(last)>0
+        last_uint = Base.CodeUnits(last)
+        last_len = length(last_uint)
+        _add_last_for_join!(init0, curr_pos, ds, last_uint, last_len, threads)
     else
-        idx = Ref{Int}(0)
-        mapreduce(identity, (x,y) -> _op_for_join!(x, y, delim, last, length(colsidx), idx, 1, length(x)), view(_columns(ds),colsidx), init = init0)
+        curr_pos .-= 1
     end
-    init0
+    res = Vector{Union{String, Missing}}(undef, nrow(ds))
+    _convert_uint8_to_string!(res, init0, curr_pos, ds, threads)
+    res
+
 end
 
 

src/byrow/util.jl

@@ -63,44 +63,50 @@ function hp_row_all_multi(ds::AbstractDataset, f::Vector{<:Function}, cols = :)
     mapreduce_index(f, _hp_op_for_all_multi!, view(_columns(ds),colsidx), ones(Bool, size(ds,1)))
 end
 
-STRING(x) = string(x)
-STRING(::Missing) = ""
-STRING(x::Bool) = x ? "1" : "0"
-STRING(::Nothing) = ""
-
-function _op_for_row_join(x, y, f, delim, quotechar, idx, p)
-    idx[] += 1
-    if quotechar === nothing
-        if idx[] < p
-            x .*= STRING.(f[idx[]].(y))
-            x .*= delim
-        else
-            x .*= STRING.(f[idx[]].(y))
-            x .*= '\n'
-        end
-    else
-        if idx[] < p
-            x .*= quotechar
-            x .*= STRING.(f[idx[]].(y))
-            x .*= quotechar
-            x .*= delim
-        else
-            x .*= quotechar
-            x .*= STRING.(f[idx[]].(y))
-            x .*= quotechar
-            x .*= '\n'
-        end
-    end
-    x
-end
-
-function row_join(ds::AbstractDataset, f::Vector{<:Function}, cols  = :; delim = ',', quotechar = nothing)
-    colsidx = index(ds)[cols]
-    idx = Ref{Int}(0)
-    p = length(colsidx)
-    init0 = fill("", nrow(ds))
-    mapreduce(identity, (x,y)->_op_for_row_join(x,y,f, delim, quotechar, idx, p), view(_columns(ds), colsidx), init = init0)
-end
+__STRING(x) = string(x)
+__STRING(x::AbstractString) = x
+__STRING(::Missing) = ""
+__STRING(x::Bool) = x ? "1" : "0"
+__STRING(::Nothing) = ""
+
+__CODEUNIT(x) = Base.CodeUnits(__STRING(x))
+
+
+
+#
+# function _op_for_row_join(x, y, f, delim, quotechar, idx, p)
+#     idx[] += 1
+#     if quotechar === nothing
+#         if idx[] < p
+#             x .*= STRING.(f[idx[]].(y))
+#             x .*= delim
+#         else
+#             x .*= STRING.(f[idx[]].(y))
+#             x .*= '\n'
+#         end
+#     else
+#         if idx[] < p
+#             x .*= quotechar
+#             x .*= STRING.(f[idx[]].(y))
+#             x .*= quotechar
+#             x .*= delim
+#         else
+#             x .*= quotechar
+#             x .*= STRING.(f[idx[]].(y))
+#             x .*= quotechar
+#             x .*= '\n'
+#         end
+#     end
+#     x
+# end
+#
+# function row_join(ds::AbstractDataset, f::Vector{<:Function}, cols  = :; delim = ',', quotechar = nothing)
+#     colsidx = index(ds)[cols]
+#     idx = Ref{Int}(0)
+#     p = length(colsidx)
+#     init0 = fill("", nrow(ds))
+#     mapreduce(identity, (x,y)->_op_for_row_join(x,y,f, delim, quotechar, idx, p), view(_columns(ds), colsidx), init = init0)
+# end
 
 # some experimental ideas
 
@@ -255,10 +261,9 @@ end
 # buffer = Matrix{UInt8}(undef, lsize , nrow(ds))
 # currentpos = ones(Int, nrow(ds))
 function row_join!(buffer, currentpos, ds::AbstractDataset, f::Vector{<:Function}, cols = :; delim = ',', quotechar = nothing, threads = true)
-    colsidx = index(ds)[cols]
+    colsidx = multiple_getindex(index(ds), cols)
     p = length(colsidx)
     dlm = UInt8.(delim)
-
     if threads
         cz = div(nrow(ds), __NCORES)
         idx = [Ref{Int}(0) for _ in 1:__NCORES]