add a new keyword argument to update/!

the `op` keyword argument is for passing a user defined function for updating rows in the main data set

Author: sl-solution

docs/src/man/joins.md

@@ -447,7 +447,9 @@ julia> contains(dsl, dsr, on = [1=>1, 2=>(2,3)], strict_inequality = true)
 
 ## Update a data set by values from another data set
 
-`update!` updates a data set values by using values from a transaction data set. The function uses the given keys (`on = ...`) to select rows for updating. By default, the missing values in transaction data set wouldn't replace the values in the main data set, however, using `allowmissing = true` changes this behaviour. If there are multiple rows in the main data set which match the key(s), using `mode = :all` causes all of them to be updated, `mode = :missing` causes only the ones which are missing in the main data set to be updated, and `mode = fun` updates the values which calling `fun` on them returns `true`. If there are multiple rows in the transaction data set which match the key, only the last one (given `stable = true` is passed) will be used to update the main data set.
+`update!` updates a data set values by using values from a transaction data set. The function uses the given keys (`on = ...`) to select rows for updating. By default, the missing values in transaction data set wouldn't replace the values in the main data set, however, using `allowmissing = true` changes this behaviour. If there are multiple rows in the main data set which match the key(s), using `mode = :all` causes all of them to be updated, `mode = :missings` causes only the ones which are missing in the main data set to be updated, and `mode = fun` updates the values which calling `fun` on them returns `true`. If there are multiple rows in the transaction data set which match the key, only the last one (given `stable = true` is passed) will be used to update the main data set.
+
+By default, `update!` updates the old values by the new values from the transaction data set, however, user may pass any function via the `op` keyword argument to update the values in the main data set by the result of calling `op` on values on both data sets. In this case, `update!` updates values in the main data set by `op(old, new)`, where `old` is the value from the main data set and `new` is the value from the transaction data set.
 
 The `update!` functions replace the main data set with the updated version, however, if a copy of the updated data set is required, the `update` function can be used instead.
 
@@ -486,7 +488,7 @@ julia> transaction = Dataset(group = ["G1", "G2"], id = [2, 1],
 
 
 julia> update(main, transaction, on = [:group, :id],
-               allowmissing = false, mode = :missing)
+               allowmissing = false, mode = :missings)
 7×4 Dataset
  Row │ group        id        x1        x2
      │ identity     identity  identity  identity
@@ -530,6 +532,20 @@ julia> update(main, transaction, on = [:group, :id],
    5 │ G2               1        1.3         1
    6 │ G2               1        2.1   missing
    7 │ G2               2        0.0         2
+
+julia> update(main, transaction, on = [:group, :id], op = +) # add values of transaction to main, when op is set mode = :all is default
+7×4 Dataset
+ Row │ group     id        x1         x2       
+     │ identity  identity  identity   identity
+     │ String?   Int64?    Float64?   Int64?   
+─────┼─────────────────────────────────────────
+   1 │ G1               1        1.2         5
+   2 │ G1               1        2.3         4
+   3 │ G1               2  missing           4
+   4 │ G1               2        4.8         2
+   5 │ G2               1        1.3         4
+   6 │ G2               1        2.1   missing
+   7 │ G2               2        0.0         2
 ```
 
 ## `compare`

src/join/join_dict.jl

@@ -503,7 +503,7 @@ function _join_outer_dict(dsl, dsr, ranges, onleft, onright, oncols_left, oncols
 
 end
 
-function _update!_dict(dsl, dsr, ranges, onleft, onright, right_cols, ::Val{T}; allowmissing = true, mode = :all, mapformats = [true, true], stable = false, alg = HeapSort, threads = threads) where T
+function _update!_dict(dsl, dsr, ranges, onleft, onright, right_cols, ::Val{T}; allowmissing = true, mode = :all, mapformats = [true, true], stable = false, alg = HeapSort, threads = threads, op = nothing) where T
     _fl = _date_value∘identity
     _fr = _date_value∘identity
     if mapformats[1]
@@ -534,7 +534,7 @@ function _update!_dict(dsl, dsr, ranges, onleft, onright, right_cols, ::Val{T};
             TL = nonmissingtype(eltype(_columns(dsl)[left_cols_idx]))
             TR = nonmissingtype(eltype(_columns(dsr)[right_cols[j]]))
             if promote_type(TR, TL) <: TL
-                _update_left_with_right!(_columns(dsl)[left_cols_idx], _columns(dsr)[right_cols[j]], ranges, allowmissing, f_mode, threads = threads)
+                _update_left_with_right!(_columns(dsl)[left_cols_idx], _columns(dsr)[right_cols[j]], ranges, allowmissing, f_mode, threads = threads, op = op)
             end
         end
     end

src/join/main.jl

@@ -1482,7 +1482,7 @@ function closejoin!(dsl::Dataset, dsr::AbstractDataset; on = nothing, direction
 end
 
 """
-    update!(dsmain::Dataset, dsupdate::AbstractDataset; on=nothing, allowmissing=false, mode=:missings, mapformats=true, alg=HeapSort, stable=true, accelerate = false, method = :sort, threads = true)
+    update!(dsmain::Dataset, dsupdate::AbstractDataset; on=nothing, allowmissing=false, mode=:missings, op = nothing, mapformats=true, alg=HeapSort, stable=true, accelerate = false, method = :sort, threads = true)
 
 Update a `Dataset` `dsmain` with another `Dataset` `dsupdate` based `on` given keys for matching rows,
 and change the left `Dataset` after updating.
@@ -1498,8 +1498,9 @@ the order of selected observation from the right table.
 - `on`: can be a single column name, a vector of column names or a vector of pairs of column names, known as keys that the update function will based on.
 - `allowmissing`: is set to `false` by default, so `missing` values in `dsupdate` will not replace the values in `dsmain`;
   change this to `true` can update `dsmain` using `missing` values in `dsupdate`.
-- `mode`: by default is set to `:missings`, means that only rows in `dsmain` with `missing` values will be updated.
+- `mode`: by default is set to `:missings` and when `op` is passed the default is set to `:all`, it means when `op` is not set only rows in `dsmain` with `missing` values will be updated.
     changing it to `:all` means all matching rows based `on` keys will be updated. Otherwise a function can be passed as `mode` to update only observations which return true when `mode` call on them.
+- `op`: by default, `update!` replace the values in `dsmain` by the values from `dsupdate`, however, user can pass any binary function to `op` to replace the value in `dsmain` by `op(left_value, right_value)`, i.e. replace it by calling `op` on the old value and the new value.
 $_JOINMAPFORMATSDOC
 $_JOINTHREADSDOC
 $_JOINMETHODDOCSORT
@@ -1552,12 +1553,33 @@ julia> update!(dsmain, dsupdate, on = [:group, :id], mode = :missings) # Only mi
    5 │ G2               1       1.3         1
    6 │ G2               1       2.1         3
    7 │ G2               2       0.0         2
+
+julia> dsmain = Dataset(group = ["G1", "G1", "G1", "G1", "G2", "G2", "G2"],
+                             id    = [ 1  ,  1  ,  2  ,  2  ,  1  ,  1  ,  2  ],
+                             x1    = [1.2, 2.3,missing,  2.3, 1.3, 2.1  , 0.0 ],
+                             x2    = [ 5  ,  4  ,  4  ,  2  , 1  ,missing, 2  ]);
+julia> dsupdate = Dataset(group = ["G1", "G2"], id = [2, 1],
+                               x1 = [2.5, missing], x2 = [missing, 3]);
+
+julia> update!(dsmain, dsupdate, on = [:group, :id], op = +, mode = :all)
+7×4 Dataset
+ Row │ group     id        x1         x2
+     │ identity  identity  identity   identity
+     │ String?   Int64?    Float64?   Int64?
+─────┼─────────────────────────────────────────
+   1 │ G1               1        1.2         5
+   2 │ G1               1        2.3         4
+   3 │ G1               2  missing           4
+   4 │ G1               2        4.8         2
+   5 │ G2               1        1.3         4
+   6 │ G2               1        2.1   missing
+   7 │ G2               2        0.0         2
 ```
 """
-function update!(dsmain::Dataset, dsupdate::AbstractDataset; on = nothing, allowmissing = false, mode::Union{Symbol, Function} = :missings,  mapformats::Union{Bool, Vector{Bool}} = true, stable = true, alg = HeapSort, accelerate = false, method = :sort, threads::Bool = true)
+function update!(dsmain::Dataset, dsupdate::AbstractDataset; on = nothing, allowmissing = false, op = nothing, mode::Union{Symbol, Function} = op === nothing ? :missings : :all,  mapformats::Union{Bool, Vector{Bool}} = true, stable = true, alg = HeapSort, accelerate = false, method = :sort, threads::Bool = true)
     !(method in (:hash, :sort)) && throw(ArgumentError("method must be :hash or :sort"))
     on === nothing && throw(ArgumentError("`on` keyword must be specified"))
-    mode isa Symbol && !(mode ∈ (:all, :missing, :missings))  && throw(ArgumentError("`mode` can be either :all or :missing"))
+    mode isa Symbol && !(mode ∈ (:all, :missing, :missings))  && throw(ArgumentError("`mode` can be either :all, :missing, or a function"))
     if !(on isa AbstractVector)
         on = [on]
     else
@@ -1577,7 +1599,7 @@ function update!(dsmain::Dataset, dsupdate::AbstractDataset; on = nothing, allow
     else
         throw(ArgumentError("`on` keyword must be a vector of column names or a vector of pairs of column names"))
     end
-    _update!(dsmain, dsupdate, nrow(dsupdate) < typemax(Int32) ? Val(Int32) : Val(Int64), onleft = onleft, onright = onright, allowmissing = allowmissing, mode = mode, mapformats = mapformats, stable = stable, alg = alg, accelerate = accelerate, method = method, threads = threads)
+    _update!(dsmain, dsupdate, nrow(dsupdate) < typemax(Int32) ? Val(Int32) : Val(Int64), onleft = onleft, onright = onright, allowmissing = allowmissing, mode = mode, mapformats = mapformats, stable = stable, alg = alg, accelerate = accelerate, method = method, threads = threads, op = op)
 
     dsmain
 end
@@ -1586,8 +1608,7 @@ end
 
 Variant of `update!` that returns an updated copy of `dsmain` leaving `dsmain` itself unmodified.
 """
-update(dsmain::AbstractDataset, dsupdate::AbstractDataset; on = nothing, allowmissing = false, mode = :all,  mapformats::Union{Bool, Vector{Bool}} = true, stable = true, alg = HeapSort, accelerate = false, method = :sort, threads = true) = update!(copy(dsmain), dsupdate; on = on, allowmissing = allowmissing, mode = mode,  mapformats = mapformats, stable = stable, alg = alg, accelerate = accelerate, method = method, threads = threads)
-
+update(dsmain::AbstractDataset, dsupdate::AbstractDataset; on = nothing, allowmissing = false, op = nothing, mode = op === nothing ? :missings : :all,  mapformats::Union{Bool, Vector{Bool}} = true, stable = true, alg = HeapSort, accelerate = false, method = :sort, threads = true,) = update!(copy(dsmain), dsupdate; on = on, allowmissing = allowmissing, mode = mode, mapformats = mapformats, stable = stable, alg = alg, accelerate = accelerate, method = method, threads = threads, op = op)
 
 
 # TODO the docstring is very limited, we need a more comprehensive docs here / and more examples

src/join/update.jl

@@ -1,18 +1,32 @@
-function _update_left_with_right!(x, y, ranges, allowmissing, mode::F; threads = true) where F
-    @_threadsfor threads for i in 1:length(x)
-        if length(ranges[i]) > 0
-            if mode(x[i])
-                if !allowmissing && !ismissing(y[ranges[i].stop])
-                    x[i] = y[ranges[i].stop]
-                elseif allowmissing
-                    x[i] = y[ranges[i].stop]
+function _update_left_with_right!(x, y, ranges, allowmissing, mode::F; threads = true, op = nothing) where F
+    if op === nothing
+        @_threadsfor threads for i in 1:length(x)
+            if length(ranges[i]) > 0
+                if mode(x[i])
+                    if !allowmissing && !ismissing(y[ranges[i].stop])
+                        x[i] = y[ranges[i].stop]
+                    elseif allowmissing
+                        x[i] = y[ranges[i].stop]
+                    end
+                end
+            end
+        end
+    else
+        @_threadsfor threads for i in 1:length(x)
+            if length(ranges[i]) > 0
+                if mode(x[i])
+                    if !allowmissing && !ismissing(y[ranges[i].stop])
+                        x[i] = op(x[i], y[ranges[i].stop])
+                    elseif allowmissing
+                        x[i] = op(x[i], y[ranges[i].stop])
+                    end
                 end
             end
         end
     end
 end
 
-function _update!(dsl::Dataset, dsr::AbstractDataset, ::Val{T}; onleft, onright, check = true, allowmissing = true, mode = :all, mapformats = [true, true], stable = false, alg = HeapSort, accelerate = false, usehash = true, method = :sort, threads = true) where T
+function _update!(dsl::Dataset, dsr::AbstractDataset, ::Val{T}; onleft, onright, check = true, allowmissing = true, mode = :all, mapformats = [true, true], stable = false, alg = HeapSort, accelerate = false, usehash = true, method = :sort, threads = true, op = nothing) where T
     isempty(dsl) && return dsl
     if method == :hash
         ranges, a, idx, minval, reps, sz, right_cols = _find_ranges_for_join_using_hash(dsl, dsr, onleft, onright, mapformats, true, Val(T); threads = threads)
@@ -23,7 +37,7 @@ function _update!(dsl::Dataset, dsr::AbstractDataset, ::Val{T}; onleft, onright,
 
         ranges = Vector{UnitRange{T}}(undef, nrow(dsl))
         if usehash && length(oncols_left) == 1 && nrow(dsr)>1
-            success, result = _update!_dict(dsl, dsr, ranges, oncols_left, oncols_right, right_cols, Val(T); mapformats = mapformats, allowmissing = allowmissing, mode = mode, threads = threads)
+            success, result = _update!_dict(dsl, dsr, ranges, oncols_left, oncols_right, right_cols, Val(T); mapformats = mapformats, allowmissing = allowmissing, mode = mode, threads = threads, op = op)
             if success
                 return result
             end
@@ -48,7 +62,7 @@ function _update!(dsl::Dataset, dsr::AbstractDataset, ::Val{T}; onleft, onright,
             TL = nonmissingtype(eltype(_columns(dsl)[left_cols_idx]))
             TR = nonmissingtype(eltype(_columns(dsr)[right_cols[j]]))
             if promote_type(TR, TL) <: TL
-                _update_left_with_right!(_columns(dsl)[left_cols_idx], view(_columns(dsr)[right_cols[j]], idx), ranges, allowmissing, f_mode, threads = threads)
+                _update_left_with_right!(_columns(dsl)[left_cols_idx], view(_columns(dsr)[right_cols[j]], idx), ranges, allowmissing, f_mode, threads = threads, op = op)
             end
         end
     end