all joins now support method = :sort or :hash

Author: sl-solution

src/join/closejoin.jl

@@ -308,7 +308,7 @@ end
 
 
 # border = :nearest | :missing | :none
-function _join_closejoin(dsl, dsr::AbstractDataset, ::Val{T}; onleft, onright, makeunique = false, border = :nearest, mapformats = [true, true], stable = false, alg = HeapSort, accelerate = false, direction = :backward, inplace = false, tol = nothing,  allow_exact_match = true, op = nothing) where T
+function _join_closejoin(dsl, dsr::AbstractDataset, ::Val{T}; onleft, onright, makeunique = false, border = :nearest, mapformats = [true, true], stable = false, alg = HeapSort, accelerate = false, direction = :backward, inplace = false, tol = nothing,  allow_exact_match = true, op = nothing, method = :sort) where T
     isempty(dsl) && return copy(dsl)
     if !allow_exact_match
         #aem is the function to check allow_exact_match
@@ -332,14 +332,24 @@ function _join_closejoin(dsl, dsr::AbstractDataset, ::Val{T}; onleft, onright, m
     if DataAPI.refpool(_columns(dsr)[oncols_right[end]]) !== nothing
         nsfpaj = false
     end
-    ranges = Vector{UnitRange{T}}(undef, nrow(dsl))
-    idx, uniquemode = _find_permute_and_fill_range_for_join!(ranges, dsr, dsl, oncols_right, oncols_left, stable, alg, mapformats, accelerate && length(oncols_right) > 1; nsfpaj = nsfpaj)
+    if length(oncols_left) > 1 && method == :hash
+        ranges, a, idx, minval, reps, sz, right_cols_2= _find_ranges_for_join_using_hash(dsl, dsr, onleft[1:end-1], onright[1:end-1], mapformats, true, Val(T))
+        filter!(!=(0), reps)
+        pushfirst!(reps, 1)
+        cumsum!(reps, reps)
+        pop!(reps)
+        grng = GIVENRANGE(idx, reps, Int[], length(reps))
+        starts, idx, last_valid_range = _sort_for_join_after_hash(dsr, oncols_right[end], stable, alg, mapformats, nsfpaj, grng)
+    else
+        ranges = Vector{UnitRange{T}}(undef, nrow(dsl))
+        idx, uniquemode = _find_permute_and_fill_range_for_join!(ranges, dsr, dsl, oncols_right, oncols_left, stable, alg, mapformats, accelerate && length(oncols_right) > 1; nsfpaj = nsfpaj)
 
-    for j in 1:(length(oncols_left) - 1)
-        _change_refpool_find_range_for_join!(ranges, dsl, dsr, idx, oncols_left, oncols_right, mapformats[1], mapformats[2], j; nsfpaj = nsfpaj)
+        for j in 1:(length(oncols_left) - 1)
+            _change_refpool_find_range_for_join!(ranges, dsl, dsr, idx, oncols_left, oncols_right, mapformats[1], mapformats[2], j; nsfpaj = nsfpaj)
+        end
     end
 
-    # if border = :none , we should :nearest direction
+    # if border = :none , we should use :nearest direction
     _change_refpool_find_range_for_close!(ranges, dsl, dsr, idx, oncols_left, oncols_right, border == :none ? :nearest : direction, mapformats[1], mapformats[2], length(oncols_left); nsfpaj = nsfpaj)
     total_length = nrow(dsl)
 

src/join/join.jl

@@ -155,6 +155,42 @@ function _fill_ranges_for_dict_join!(ranges, dict, maxprob, _fl, _fr, x_l, x_r,
 end
 
 
+function _find_ranges_for_join_using_hash(dsl, dsr, onleft, onright, mapformats, makeunique, ::Val{T}) where T
+    oncols_left = onleft
+    oncols_right = onright
+    right_cols = setdiff(1:length(index(dsr)), oncols_right)
+    if !makeunique && !isempty(intersect(_names(dsl), _names(dsr)[right_cols]))
+        throw(ArgumentError("duplicate column names, pass `makeunique = true` to make them unique using a suffix automatically." ))
+    end
+
+    cols = Any[]
+    for j in 1:length(oncols_left)
+        if mapformats[1]
+            fl = getformat(dsl, oncols_left[j])
+        else
+            fl = identity
+        end
+        if mapformats[2]
+            fr = getformat(dsr, oncols_right[j])
+        else
+            fr = identity
+        end
+        push!(cols, Cat2Vec(_columns(dsl)[oncols_left[j]], _columns(dsr)[oncols_right[j]], fl, fr))
+    end
+    newds = Dataset(cols, :auto, copycols = false)
+    a = _gather_groups(newds, :, nrow(newds)< typemax(Int32) ? Val(Int32) : Val(Int64), stable = false, mapformats = false)
+
+    reps = _find_counts_for_join(view(a[1], nrow(dsl)+1:length(a[1])), a[3])
+    gslots, minval, sz = _create_dictionary_for_join_int(identity, view(a[1], nrow(dsl)+1:length(a[1])), reps, 1, a[3], Val(T))
+
+    ranges = Vector{UnitRange{T}}(undef, nrow(dsl))
+    where = Vector{T}(undef, length(reps)+1)
+    cumsum!(view(where, 2:length(where)), reps)
+    where[1] = 0
+    _find_range_for_join!(ranges, view(a[1], 1:nrow(dsl)), gslots, reps, where, 1, sz)
+    ranges, a, gslots, minval, reps, sz, right_cols
+end
+
 function _join_left_dict(dsl, dsr, ranges, onleft, onright, right_cols, ::Val{T}; makeunique = makeunique, mapformats = mapformats, check = check ) where T
     _fl = _date_value∘identity
     _fr = _date_value∘identity
@@ -231,7 +267,9 @@ function _join_left!_dict(dsl, dsr, ranges, onleft, onright, right_cols, ::Val{T
     end
 
     _fill_ranges_for_dict_join!(ranges, dict, maxprob, _fl, _fr, _columns(dsl)[onleft[1]], _columns(dsr)[onright[1]], sz, type)
-
+    if !all(x->length(x) <= 1, ranges)
+        throw(ArgumentError("`leftjoin!` can only be used when each observation in left data set matches at most one observation from right data set"))
+    end
     new_ends = map(x -> max(1, length(x)), ranges)
     cumsum!(new_ends, new_ends)
     total_length = new_ends[end]
@@ -241,10 +279,10 @@ function _join_left!_dict(dsl, dsr, ranges, onleft, onright, right_cols, ::Val{T
     end
 
     for j in 1:length(right_cols)
-        _res = allocatecol(_columns(dsr)[right_cols[j]], total_length, addmissing = false)
+        _res = allocatecol(_columns(dsr)[right_cols[j]], total_length)
         if DataAPI.refpool(_res) !== nothing
-            # fill_val = DataAPI.invrefpool(_res)[missing]
-            _fill_right_cols_table_left!(_res.refs, DataAPI.refarray(_columns(dsr)[right_cols[j]]), ranges, new_ends, total_length, missing)
+            fill_val = DataAPI.invrefpool(_res)[missing]
+            _fill_right_cols_table_left!(_res.refs, DataAPI.refarray(_columns(dsr)[right_cols[j]]), ranges, new_ends, total_length, fill_val)
         else
             _fill_right_cols_table_left!(_res, _columns(dsr)[right_cols[j]], ranges, new_ends, total_length, missing)
         end
@@ -376,16 +414,6 @@ function _join_outer_dict(dsl, dsr, ranges, onleft, onright, oncols_left, oncols
 
 end
 
-function _in_use_Set(ldata, rdata, _fl, _fr)
-    ss = Set(Base.Generator(_fr, rdata));
-    res = Vector{Bool}(undef, length(ldata))
-    Threads.@threads for i in 1:length(res)
-        res[i] = _fl(ldata[i]) in ss
-    end
-    res
-end
-
-
 function _update!_dict(dsl, dsr, ranges, onleft, onright, right_cols, ::Val{T}; allowmissing = true, mode = :all, mapformats = [true, true], stable = false, alg = HeapSort) where T
     _fl = _date_value∘identity
     _fr = _date_value∘identity
@@ -424,3 +452,128 @@ function _update!_dict(dsl, dsr, ranges, onleft, onright, right_cols, ::Val{T};
     _modified(_attributes(dsl))
     true, dsl
 end
+
+
+# a new idea for joining without sorting
+function _in_hash(dsl::AbstractDataset, dsr::AbstractDataset, ::Val{T}; onleft, onright, mapformats = [true, true]) where T
+    isempty(dsl) && return Bool[]
+    oncols_left = onleft
+    oncols_right = onright
+
+    # use Set when there is only one column in `on`
+    cols = Any[]
+    for j in 1:length(oncols_left)
+        if mapformats[1]
+            fl = getformat(dsl, oncols_left[j])
+        else
+            fl = identity
+        end
+        if mapformats[2]
+            fr = getformat(dsr, oncols_right[j])
+        else
+            fr = identity
+        end
+        push!(cols, Cat2Vec(_columns(dsl)[oncols_left[j]], _columns(dsr)[oncols_right[j]], fl, fr))
+    end
+    newds = Dataset(cols, :auto, copycols = false)
+    a = _gather_groups(newds, :, nrow(newds)< typemax(Int32) ? Val(Int32) : Val(Int64), stable = false, mapformats = false)
+    res = _in_use_Set_int(view(a[1], 1:nrow(dsl)), view(a[1], nrow(dsl)+1:length(a[1])), 1, a[3])
+end
+
+function _create_Set_for_join_int(f, v, minval, rangelen)
+    flag = false
+    offset = 1 - minval
+    n = length(v)
+    sz = rangelen + 1
+    gslots = falses(sz)
+    @inbounds for i in 1:length(v)
+        _fv = f(v[i])
+        if ismissing(_fv)
+            slotix = sz
+        else
+            slotix = _fv + offset
+        end
+        if !gslots[slotix]
+            gslots[slotix] = true
+        end
+    end
+    gslots, minval, sz
+end
+
+function _query_Set_for_join_int(f, fv, gslots, minval, sz)
+    offset = 1 - minval
+    if ismissing(fv)
+        slotix = sz
+    else
+        slotix = fv + offset
+        !(slotix in 1:sz-1) && return 0
+    end
+
+    if slotix in 1:sz
+        rowid = gslots[slotix]
+        return rowid
+    end
+    false
+end
+
+function _in_use_Set_int_barrier!(res, ldata, gslots, minval, sz)
+    Threads.@threads for i in 1:length(res)
+        res[i] = _query_Set_for_join_int(identity, ldata[i], gslots, minval, sz)
+    end
+end
+
+function _in_use_Set_int(ldata, rdata, minval, rangelen)
+    gslots, minval, sz  =  _create_Set_for_join_int(identity, rdata, minval, rangelen)
+    res = Vector{Bool}(undef, length(ldata))
+    _in_use_Set_int_barrier!(res, ldata, gslots, minval, sz)
+    res
+end
+
+# f is a function which should be applied on each element of v
+# v is a vector of Int with minimum minval and range length of rangelen
+# reps gives how many times a specific integer will appear in v
+# no missing in v
+function _create_dictionary_for_join_int(f, v, reps, minval, rangelen, ::Val{T}) where T
+    offset = 1 - minval
+    n = length(v)
+    where = Vector{T}(undef, length(reps)+1)
+    cumsum!(view(where, 2:length(where)), reps)
+    where[1] = 0
+    gslots = zeros(T, where[end])
+    sz = rangelen
+    @inbounds for i in 1:length(v)
+        _fv = f(v[i])
+        slotix = _fv + offset
+        gslots[where[slotix]+1] = i
+        where[slotix] += 1
+    end
+    gslots, minval, sz
+end
+# there is no missing in v
+# here can be defined as:
+# where = Vector{T}(undef, length(reps)+1)
+# cumsum!(view(where, 2:length(where)), reps)
+# where[1] = 0
+function _query_dictionary_for_join_int(f, v, gslots,reps, where, minval, sz)
+    offset = 1 - minval
+    slotix = v + offset
+    !(slotix in 1:sz) && return 1:0
+    if slotix in 1:sz
+        rowid = reps[slotix]
+        rowid == 0 && return 1:0
+        return where[slotix]+1:where[slotix+1]
+    end
+end
+
+
+
+
+function _in_use_Set(ldata, rdata, _fl, _fr)
+
+    ss = Set(Base.Generator(_fr, rdata));
+    res = Vector{Bool}(undef, length(ldata))
+    Threads.@threads for i in 1:length(res)
+        res[i] = _fl(ldata[i]) in ss
+    end
+    res
+end