_add_sum(x, y) = Base.add_sum(x, y)

_add_sum(x, ::Missing) = x

_add_sum(::Missing, x) = x

_add_sum(::Missing, ::Missing) = missing

_mul_prod(x, y) = Base.mul_prod(x, y)

_mul_prod(x, ::Missing) = x

_mul_prod(::Missing, x) = x

_mul_prod(::Missing, ::Missing) = missing

_min_fun(x, y) = min(x, y)

_min_fun(x, ::Missing) = x

_min_fun(::Missing, y) = y

_min_fun(::Missing, ::Missing) = missing

_max_fun(x, y) = max(x, y)

_max_fun(x, ::Missing) = x

_max_fun(::Missing, y) = y

_max_fun(::Missing, ::Missing) = missing

_bool(f) = x->f(x)::Bool

struct _Prehashed

hash::UInt64

end

Base.hash(x::_Prehashed) = x.hash

Base.@propagate_inbounds function _op_for_sum!(x, y, f, lo, hi)

@simd for i in lo:hi

x[i] = _add_sum(x[i], f(y[i]))

end

x

end

function row_sum(ds::AbstractDataset, f::Function, cols = names(ds, Union{Missing, Number}); threads = true)

colsidx = multiple_getindex(index(ds), cols)

CT = mapreduce(eltype, promote_type, view(_columns(ds),colsidx))

T = Core.Compiler.return_type(f, Tuple{CT})

CT = nonmissingtype(T)

CT <: Base.SmallSigned ? CT = Int : nothing

CT <: Base.SmallUnsigned ? CT = UInt : nothing

CT <: Bool ? CT = Int : nothing

T = Union{Missing, CT}

init0 = _missings(T, nrow(ds))

if threads

cz = div(length(init0), Threads.nthreads())

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

mapreduce(identity, (x,y) -> _op_for_sum!(x,y, f, lo, hi), view(_columns(ds),colsidx), init = init0)

end

else

mapreduce(identity, (x,y) -> _op_for_sum!(x,y, f, 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

init0

end

row_sum(ds::AbstractDataset, cols = names(ds, Union{Missing, Number}); threads = true) = row_sum(ds, identity, cols, threads = threads)

Base.@propagate_inbounds function _op_for_prod!(x, y, f, lo, hi)

@simd for i in lo:hi

x[i] = _mul_prod(x[i], f(y[i]))

end

x

end

function row_prod(ds::AbstractDataset, f::Function, cols = names(ds, Union{Missing, Number}); threads = true)

colsidx = multiple_getindex(index(ds), cols)

CT = mapreduce(eltype, promote_type, view(_columns(ds),colsidx))

T = Core.Compiler.return_type(f, Tuple{CT})

CT = nonmissingtype(T)

CT <: Base.SmallSigned ? CT = Int : nothing

CT <: Base.SmallUnsigned ? CT = UInt : nothing

CT <: Bool ? CT = Int : nothing

T = Union{Missing, CT}

init0 = _missings(T, nrow(ds))

if threads

cz = div(length(init0), Threads.nthreads())

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

mapreduce(identity, (x,y) -> _op_for_prod!(x,y, f, lo, hi), view(_columns(ds),colsidx), init = init0)

end

else

mapreduce(identity, (x,y) -> _op_for_prod!(x,y, f, 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

init0

end

row_prod(ds::AbstractDataset, cols = names(ds, Union{Missing, Number}); threads = true) = row_prod(ds, identity, cols; threads = threads)

Base.@propagate_inbounds function _op_for_count!(x, y, f, lo, hi)

@simd for i in lo:hi

x[i] += f(y[i])

end

x

end

function row_count(ds::AbstractDataset, f::Function, cols = names(ds, Union{Missing, Number}); threads = true)

colsidx = multiple_getindex(index(ds), cols)

init0 = zeros(Int32, size(ds,1))

if threads

cz = div(length(init0), Threads.nthreads())

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

mapreduce(identity, (x,y) -> _op_for_count!(x, y, _bool(f), lo, hi), view(_columns(ds),colsidx), init = init0)

end

else

mapreduce(identity, (x,y) -> _op_for_count!(x, y, _bool(f), 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

init0

end

row_count(ds::AbstractDataset, cols = names(ds, Union{Missing, Number}); threads = true) = row_count(ds, x->true, cols; threads = threads)

_op_bool_add(x::Bool,y::Bool) = x | y ? true : false

Base.@propagate_inbounds function op_for_any!(x, y, f, lo, hi)

@simd for i in lo:hi

!x[i] ? x[i] = f(y[i]) : nothing

end

x

end

function row_any(ds::AbstractDataset, f::Union{AbstractVector{<:Function}, Function}, cols = :; threads = true)

colsidx = multiple_getindex(index(ds), cols)

init0 = zeros(Bool, size(ds,1))

multi_f = false

if f isa AbstractVector

@assert length(f) == length(colsidx) "number of provided functions must match the number of selected columns"

multi_f = true

end

if threads

cz = div(length(init0), Threads.nthreads())

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

if multi_f

mapreduce_index(f, (x, y, func) -> op_for_any!(x, y, _bool(func), lo, hi), view(_columns(ds),colsidx), init0)

else

mapreduce(identity, (x,y) -> op_for_any!(x,y, _bool(f), lo, hi), view(_columns(ds),colsidx), init = init0)

end

end

else

if multi_f

mapreduce_index(f, (x, y, func) -> op_for_any!(x, y, _bool(func), 1, length(x)), view(_columns(ds),colsidx), init0)

else

mapreduce(identity, (x,y) -> op_for_any!(x,y, _bool(f), 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

end

init0

end

row_any(ds::AbstractDataset, cols = :; threads = true) = row_any(ds, isequal(true), cols; threads = threads)

_op_bool_mult(x::Bool,y::Bool) = x & y ? true : false

Base.@propagate_inbounds function op_for_all!(x, y, f, lo, hi)

@simd for i in lo:hi

x[i] ? x[i] = f(y[i]) : nothing

end

x

end

function row_all(ds::AbstractDataset, f::Union{AbstractVector{<:Function}, Function}, cols = :; threads = true)

colsidx = multiple_getindex(index(ds), cols)

init0 = ones(Bool, size(ds,1))

multi_f = false

if f isa AbstractVector

@assert length(f) == length(colsidx) "number of provided functions must match the number of selected columns"

multi_f = true

end

if threads

cz = div(length(init0), Threads.nthreads())

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

if multi_f

mapreduce_index(f, (x, y, func) -> op_for_all!(x, y, _bool(func), lo, hi), view(_columns(ds),colsidx), init0)

else

mapreduce(identity, (x,y) -> op_for_all!(x,y, _bool(f), lo, hi), view(_columns(ds),colsidx), init = init0)

end

end

else

if multi_f

mapreduce_index(f, (x, y, func) -> op_for_all!(x, y, _bool(func), 1, length(x)), view(_columns(ds),colsidx), init0)

else

mapreduce(identity, (x,y) -> op_for_all!(x,y, _bool(f), 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

end

init0

end

row_all(ds::AbstractDataset, cols = :; threads = true) = row_all(ds, isequal(true), cols; threads = threads)

# this is a general rule for order of arguments in isequal, isless, findfirst, ...

# if the keyword argument is `with` then eq(y, with)

# if the keyword argument is `item` then eq(item, y)

Base.@propagate_inbounds function _op_for_isequal!(x, y, x1, lo, hi)

@simd for i in lo:hi

x[i] ? x[i] = isequal(y[i], x1[i]) : nothing

end

x

end

function row_isequal(ds::AbstractDataset, cols = :; by::Union{AbstractVector, DatasetColumn, SubDatasetColumn, ColumnIndex, Nothing} = nothing, threads = true)

colsidx = multiple_getindex(index(ds), cols)

if !(by isa ColumnIndex) && by !== nothing

@assert length(by) == nrow(ds) "to compare values of selected columns in each row, the length of the passed vector and the number of rows must match"

end

if by isa SubDatasetColumn || by isa DatasetColumn

x1 = __!(by)

elseif by isa ColumnIndex

x1 = _columns(ds)[index(ds)[by]]

elseif by === nothing

x1 = _columns(ds)[colsidx[1]]

else

x1 = by

end

init0 = ones(Bool, nrow(ds))

if threads

cz = div(length(init0), Threads.nthreads())

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

mapreduce(identity, (x,y) -> _op_for_isequal!(x,y, x1, lo, hi), view(_columns(ds),colsidx), init = init0)

end

else

mapreduce(identity, (x,y) -> _op_for_isequal!(x,y, x1, 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

init0

end

Base.@propagate_inbounds function _op_for_isless!(x, y, vals, rev,lt, lo, hi)

if !rev

@simd for i in lo:hi

x[i] ? x[i] = lt(y[i], vals[i]) : nothing

end

else

@simd for i in lo:hi

x[i] ? x[i] = lt(vals[i], y[i]) : nothing

end

end

x

end

function row_isless(ds::AbstractDataset, cols, colselector::Union{AbstractVector, DatasetColumn, SubDatasetColumn, ColumnIndex}; threads = true, rev = false, lt = isless)

if !(colselector isa ColumnIndex)

@assert length(colselector) == nrow(ds) "to compare values of selected columns in each row, the length of the passed vector and the number of rows must match"

end

colsidx = multiple_getindex(index(ds), cols)

if colselector isa SubDatasetColumn || colselector isa DatasetColumn

colselector = __!(colselector)

elseif colselector isa ColumnIndex

colselector = _columns(ds)[index(ds)[colselector]]

end

init0 = ones(Bool, nrow(ds))

if threads

cz = div(length(init0), Threads.nthreads())

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

mapreduce(identity, (x,y) -> _op_for_isless!(x, y, colselector, rev, lt, lo, hi), view(_columns(ds),colsidx), init = init0)

end

else

mapreduce(identity, (x,y) -> _op_for_isless!(x, y, colselector, rev, lt, 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

init0

end

# TODO probably we should use this approach instead of mapreduce_indexed

Base.@propagate_inbounds function _op_for_findfirst!(x, y, f, idx, missref, lo, hi)

idx[] += 1

@simd for i in lo:hi

x[i] = ifelse(isequal(missref, x[i]) && isequal(true, f(y[i])), idx[], x[i])

end

x

end

Base.@propagate_inbounds function _op_for_findfirst!(x, y, item, idx, missref, eq, lo, hi)

idx[] += 1

@simd for i in lo:hi

x[i] = ifelse(isequal(missref, x[i]) && eq(item[i], y[i]), idx[], x[i])

end

x

end

Base.@propagate_inbounds function _op_for_findlast!(x, y, f, idx, missref, lo, hi)

idx[] += 1

@simd for i in lo:hi

x[i] = ifelse(isequal(true, f(y[i])), idx[], x[i])

end

x

end

Base.@propagate_inbounds function _op_for_findlast!(x, y, item, idx, missref, eq, lo, hi)

idx[] += 1

@simd for i in lo:hi

x[i] = ifelse(eq(item[i], y[i]), idx[], x[i])

end

x

end

function row_findfirst(ds::AbstractDataset, f, cols = names(ds, Union{Missing, Number});item::Union{AbstractVector, DatasetColumn, SubDatasetColumn, ColumnIndex, Nothing} = nothing, threads = true, eq = isequal)

if !(item isa ColumnIndex) && item !== nothing

@assert length(item) == nrow(ds) "length of item values must be the same as the number of rows"

elseif item isa SubDatasetColumn || item isa DatasetColumn

item = __!(item)

elseif item isa ColumnIndex

item = _columns(ds)[index(ds)[item]]

end

colsidx = multiple_getindex(index(ds), cols)

colnames_pa = allowmissing(PooledArray(_names(ds)[colsidx]))

push!(colnames_pa, missing)

missref = get(colnames_pa.invpool, missing, 0)

init0 = fill(missref, nrow(ds))

if threads

cz = div(length(init0), Threads.nthreads())

idx = [Ref{Int}(0) for _ in 1:Threads.nthreads()]

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

if item === nothing

mapreduce(identity, (x,y) -> _op_for_findfirst!(x, y, f, idx[i], missref, lo, hi), view(_columns(ds),colsidx), init = init0)

else

mapreduce(identity, (x,y) -> _op_for_findfirst!(x, y, item, idx[i], missref, eq, lo, hi), view(_columns(ds),colsidx), init = init0)

end

end

else

idx = Ref{Int64}(0)

if item === nothing

mapreduce(identity, (x,y) -> _op_for_findfirst!(x, y, f, idx, missref, 1, length(x)), view(_columns(ds),colsidx), init = init0)

else

mapreduce(identity, (x,y) -> _op_for_findfirst!(x, y, item, idx, missref, eq, 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

end

colnames_pa.refs = init0

colnames_pa

end

function row_findlast(ds::AbstractDataset, f, cols = names(ds, Union{Missing, Number}); item::Union{AbstractVector, DatasetColumn, SubDatasetColumn, ColumnIndex, Nothing} = nothing, threads = true, eq = isequal)

if !(item isa ColumnIndex) && item !== nothing

@assert length(item) == nrow(ds) "length of item values must be the same as the number of rows"

elseif item isa SubDatasetColumn || item isa DatasetColumn

item = __!(item)

elseif item isa ColumnIndex

item = _columns(ds)[index(ds)[item]]

end

colsidx = multiple_getindex(index(ds), cols)

colnames_pa = allowmissing(PooledArray(_names(ds)[colsidx]))

push!(colnames_pa, missing)

missref = get(colnames_pa.invpool, missing, 0)

init0 = fill(missref, nrow(ds))

if threads

cz = div(length(init0), Threads.nthreads())

idx = [Ref{Int}(0) for _ in 1:Threads.nthreads()]

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

if item === nothing

mapreduce(identity, (x,y) -> _op_for_findlast!(x, y, f, idx[i], missref, lo, hi), view(_columns(ds),colsidx), init = init0)

else

mapreduce(identity, (x,y) -> _op_for_findlast!(x, y, item, idx[i], missref, eq, lo, hi), view(_columns(ds),colsidx), init = init0)

end

end

else

idx = Ref{Int}(0)

if item === nothing

mapreduce(identity, (x,y) -> _op_for_findlast!(x, y, f, idx, missref, 1, length(x)), view(_columns(ds),colsidx), init = init0)

else

mapreduce(identity, (x,y) -> _op_for_findlast!(x, y, item, idx, missref, eq, 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

end

colnames_pa.refs = init0

colnames_pa

end

Base.@propagate_inbounds function _op_for_in!(x, y, x1, eq, lo, hi)

@simd for i in lo:hi

!x[i] ? x[i] = eq(x1[i], y[i]) : nothing

end

x

end

function row_in(ds::AbstractDataset, collections, items::Union{AbstractVector, DatasetColumn, SubDatasetColumn, ColumnIndex}; threads = true, eq = isequal)

if !(items isa ColumnIndex)

@assert length(items) == nrow(ds) "length of item values must be the same as the number of rows"

end

colsidx = index(ds)[collections]

if items isa SubDatasetColumn || items isa DatasetColumn

items = __!(items)

elseif items isa ColumnIndex

items = _columns(ds)[index(ds)[items]]

end

init0 = zeros(Bool, nrow(ds))

if threads

cz = div(length(init0), Threads.nthreads())

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

mapreduce(identity, (x,y) -> _op_for_in!(x, y, items, eq, lo, hi), view(_columns(ds),colsidx), init = init0)

end

else

mapreduce(identity, (x,y) -> _op_for_in!(x, y, items, eq, 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

init0

end

Base.@propagate_inbounds function _op_for_select!(x, y, colselector, dsnames, idx, lo, hi)

idx[] += 1

@simd for i in lo:hi

if isequal(colselector[i], dsnames[idx[]])

x[i] = y[i]

end

end

x

end

function row_select(ds::AbstractDataset, cols, colselector::Union{AbstractVector, DatasetColumn, SubDatasetColumn, ColumnIndex}; threads = true)

if !(colselector isa ColumnIndex)

@assert length(colselector) == nrow(ds) "to pick values of selected columns in each row, the length of the column names and the number of rows must match, i.e. the length of the vector passed as `by` must be $(nrow(ds))."

end

colsidx = multiple_getindex(index(ds), cols)

CT = mapreduce(eltype, promote_type, view(_columns(ds),colsidx))

if colselector isa SubDatasetColumn || colselector isa DatasetColumn

colselector = __!(colselector)

end

if colselector isa ColumnIndex

colselector = _columns(ds)[index(ds)[colselector]]

end

if eltype(colselector) <: Union{Missing, Symbol}

nnames = Symbol.(names(ds, colsidx))

elseif eltype(colselector) <: Union{Missing, AbstractString}

nnames = names(ds, colsidx)

elseif eltype(colselector) <: Union{Missing, Integer}

nnames = 1:length(colsidx)

end

init0 = _missings(CT, nrow(ds))

if threads

cz = div(length(init0), Threads.nthreads())

idx = [Ref{Int}(0) for _ in 1:Threads.nthreads()]

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

mapreduce(identity, (x,y) -> _op_for_select!(x, y, colselector, nnames, idx[i], lo, hi), view(_columns(ds),colsidx), init = init0)

end

else

idx = Ref{Int}(0)

mapreduce(identity, (x,y) -> _op_for_select!(x, y, colselector, nnames, idx, 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

init0

end

Base.@propagate_inbounds function _op_for_fill!(x, y, f, lo, hi)

@simd for i in lo:hi

y[i] = ifelse(f(y[i]), x[i], y[i])

end

x

end

Base.@propagate_inbounds function _op_for_fill_roll!(x, y, f, lo, hi)

@simd for i in lo:hi

y[i] = ifelse(f(y[i]), x[i], y[i])

end

y

end

# rolling = true, means fill a column an use its value for next filling

function row_fill!(ds::AbstractDataset, cols, val::Union{AbstractVector, DatasetColumn, SubDatasetColumn, ColumnIndex}; f = ismissing, threads = true, rolling = false)

if !(val isa ColumnIndex)

@assert length(val) == nrow(ds) "to fill values in each row, the length of passed values and the number of rows must match."

end

colsidx = multiple_getindex(index(ds), cols)

if val isa SubDatasetColumn || val isa DatasetColumn

val = __!(val)

end

if val isa ColumnIndex

val = _columns(ds)[index(ds)[val]]

end

init0 = val

if threads

cz = div(length(init0), Threads.nthreads())

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

if rolling

mapreduce(identity, (x,y) -> _op_for_fill_roll!(x, y, f, lo, hi), view(_columns(ds),colsidx), init = init0)

else

mapreduce(identity, (x,y) -> _op_for_fill!(x, y, f, lo, hi), view(_columns(ds),colsidx), init = init0)

end

end

else

if rolling

mapreduce(identity, (x,y) -> _op_for_fill_roll!(x, y, f, 1, length(x)), view(_columns(ds),colsidx), init = init0)

else

mapreduce(identity, (x,y) -> _op_for_fill!(x, y, f, 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

end

any(index(parent(ds)).sortedcols .∈ Ref(IMD.parentcols.(Ref(IMD.index(ds)), cols))) && _reset_grouping_info!(parent(ds))

_modified(_attributes(parent(ds)))

ds

end

Base.@propagate_inbounds function _op_for_coalesce!(x, y, lo, hi)

if all(!ismissing, view(x, lo:hi))

x

else

@simd for i in lo:hi

x[i] = ifelse(ismissing(x[i]), y[i], x[i])

end

end

x

end

function row_coalesce(ds::AbstractDataset, cols = names(ds, Union{Missing, Number}); threads = true)

colsidx = multiple_getindex(index(ds), cols)

CT = mapreduce(eltype, promote_type, view(_columns(ds),colsidx))

init0 = _missings(CT, size(ds,1))

if threads

cz = div(length(init0), Threads.nthreads())

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

mapreduce(identity, (x,y) -> _op_for_coalesce!(x, y, lo, hi), view(_columns(ds),colsidx), init = init0)

end

else

mapreduce(identity, (x,y) -> _op_for_coalesce!(x, y, 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

init0

end

function row_mean(ds::AbstractDataset, f::Function, cols = names(ds, Union{Missing, Number}); threads = true)

row_sum(ds, f, cols; threads = threads) ./ row_count(ds, x -> !ismissing(x), cols; threads = threads)

end

row_mean(ds::AbstractDataset, cols = names(ds, Union{Missing, Number}); threads = true) = row_mean(ds, identity, cols; threads = threads)

Base.@propagate_inbounds function _op_for_min!(x, y, f, lo, hi)

@simd for i in lo:hi

x[i] = _min_fun(x[i], f(y[i]))

end

x

end

Base.@propagate_inbounds function _op_for_max!(x, y, f, lo, hi)

@simd for i in lo:hi

x[i] = _max_fun(x[i], f(y[i]))

end

x

end

function row_minimum(ds::AbstractDataset, f::Function, cols = names(ds, Union{Missing, Number}); threads = true)

colsidx = multiple_getindex(index(ds), cols)

CT = mapreduce(eltype, promote_type, view(_columns(ds),colsidx))

T = Core.Compiler.return_type(f, Tuple{CT})

init0 = _missings(T, nrow(ds))

if threads

cz = div(length(init0), Threads.nthreads())

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

mapreduce(identity, (x,y) -> _op_for_min!(x, y, f, lo, hi), view(_columns(ds),colsidx), init = init0)

end

else

mapreduce(identity, (x,y) -> _op_for_min!(x, y, f, 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

init0

end

row_minimum(ds::AbstractDataset, cols = names(ds, Union{Missing, Number}); threads = true) = row_minimum(ds, identity, cols; threads = threads)

function row_maximum(ds::AbstractDataset, f::Function, cols = names(ds, Union{Missing, Number}); threads = true)

colsidx = multiple_getindex(index(ds), cols)

CT = mapreduce(eltype, promote_type, view(_columns(ds),colsidx))

T = Core.Compiler.return_type(f, Tuple{CT})

init0 = _missings(T, nrow(ds))

if threads

cz = div(length(init0), Threads.nthreads())

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

mapreduce(identity, (x,y) -> _op_for_max!(x, y, f, lo, hi), view(_columns(ds),colsidx), init = init0)

end

else

mapreduce(identity, (x,y) -> _op_for_max!(x, y, f, 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

init0

end

row_maximum(ds::AbstractDataset, cols = names(ds, Union{Missing, Number}); threads = true) = row_maximum(ds, identity, cols; threads = threads)

Base.@propagate_inbounds function _op_for_argminmax!(x, y, f, vals, idx, missref, lo, hi)

idx[] += 1

@simd for i in lo:hi

if !ismissing(vals[i]) && isequal(vals[i], f(y[i])) && isequal(x[i], missref)

x[i] = idx[]

end

end

x

end

function row_argmin(ds::AbstractDataset, f::Function, cols = names(ds, Union{Missing, Number}); threads = true)

colsidx = multiple_getindex(index(ds), cols)

minvals = row_minimum(ds, f, cols; threads = threads)

colnames_pa = allowmissing(PooledArray(_names(ds)[colsidx]))

push!(colnames_pa, missing)

missref = get(colnames_pa.invpool, missing, missing)

init0 = fill(missref, nrow(ds))

if threads

cz = div(length(init0), Threads.nthreads())

idx = [Ref{Int}(0) for _ in 1:Threads.nthreads()]

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

mapreduce(identity, (x,y) -> _op_for_argminmax!(x, y, f, minvals, idx[i], missref, lo, hi), view(_columns(ds),colsidx), init = init0)

end

else

idx = Ref{Int}(0)

mapreduce(identity, (x,y) -> _op_for_argminmax!(x, y, f, minvals, idx, missref, 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

colnames_pa.refs = init0

colnames_pa

end

row_argmin(ds::AbstractDataset, cols = names(ds, Union{Missing, Number}); threads = true) = row_argmin(ds, identity, cols, threads = threads)

function row_argmax(ds::AbstractDataset, f::Function, cols = names(ds, Union{Missing, Number}); threads = true)

colsidx = multiple_getindex(index(ds), cols)

maxvals = row_maximum(ds, f, cols; threads = threads)

colnames_pa = allowmissing(PooledArray(_names(ds)[colsidx]))

push!(colnames_pa, missing)

missref = get(colnames_pa.invpool, missing, missing)

init0 = fill(missref, nrow(ds))

if threads

cz = div(length(init0), Threads.nthreads())

idx = [Ref{Int}(0) for _ in 1:Threads.nthreads()]

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

mapreduce(identity, (x,y) -> _op_for_argminmax!(x, y, f, maxvals, idx[i], missref, lo, hi), view(_columns(ds),colsidx), init = init0)

end

else

idx = Ref{Int}(0)

mapreduce(identity, (x,y) -> _op_for_argminmax!(x, y, f, maxvals, idx, missref, 1, length(x)), view(_columns(ds),colsidx), init = init0)

end

colnames_pa.refs = init0

colnames_pa

end

row_argmax(ds::AbstractDataset, cols = names(ds, Union{Missing, Number}); threads = true) = row_argmax(ds, identity, cols, threads = threads)

# TODO better function for the first component of operator

function _row_wise_var(ss, sval, n, dof, T)

res = Vector{T}(undef, length(ss))

for i in 1:length(ss)

if n[i] == 0

res[i] = missing

elseif n[i] == 1

res[i] = zero(T)

else

res[i] = ss[i]/n[i] - (sval[i]/n[i])*(sval[i]/n[i])

if dof

res[i] = (n[i] * res[i])/(n[i]-1)

end

end

end

res

end

# TODO needs type stability

# TODO need abs2 for calculations

function row_var(ds::AbstractDataset, f::Function, cols = names(ds, Union{Missing, Number}); dof = true, threads = true)

colsidx = multiple_getindex(index(ds), cols)

CT = mapreduce(eltype, promote_type, view(_columns(ds),colsidx))

T = Core.Compiler.return_type(f, Tuple{CT})

_sq_(x) = x^2

ss = row_sum(ds, _sq_ ∘ f, cols; threads = threads)

sval = row_sum(ds, f, cols; threads = threads)

n = row_count(ds, x -> !ismissing(x), cols; threads = threads)

T2 = Core.Compiler.return_type(/, Tuple{eltype(ss), eltype(n)})

res = Vector{Union{Missing, T2}}(undef, length(ss))

res .= ss ./ n .- (sval ./ n) .^ 2

if dof

res .= (n .* res) ./ (n .- 1)

res .= ifelse.(n .== 1, missing, res)

end

res

# _row_wise_var(ss, sval, n, dof, T)

end

row_var(ds::AbstractDataset, cols = names(ds, Union{Missing, Number}); dof = true, threads = true) = row_var(ds, identity, cols, dof = dof, threads = threads)

function row_std(ds::AbstractDataset, f::Function, cols = names(ds, Union{Missing, Number}); dof = true, threads = true)

sqrt.(row_var(ds, f, cols, dof = dof, threads = threads))

end

row_std(ds::AbstractDataset, cols = names(ds, Union{Missing, Number}); dof = true, threads = true) = row_std(ds, identity, cols, dof = dof, threads = threads)

Base.@propagate_inbounds function _op_for_cumsum_skip!(x, y, lo, hi)

@simd for i in lo:hi

x[i] = _add_sum(x[i], y[i])

end

@simd for i in lo:hi

y[i] = ifelse(ismissing(y[i]), missing, x[i])

end

x

end

Base.@propagate_inbounds function _op_for_cumsum_ignore!(x, y, lo, hi)

@simd for i in lo:hi

y[i] = _add_sum(x[i], y[i])

end

y

end

function row_cumsum!(ds::Dataset, cols = names(ds, Union{Missing, Number}); missings = :ignore, threads = true)

!(missings in (:ignore, :skip)) && throw(ArgumentError("`missings` can be either `:ignore` or `:skip`"))

colsidx = index(ds)[cols]

T = mapreduce(eltype, promote_type, view(_columns(ds),colsidx))

if T <: Union{Missing, INTEGERS}

T <: Union{Missing, Base.SmallSigned}

T = T <: Union{Missing, Base.SmallSigned, Bool} ? Union{Int, Missing} : T

T = T <: Union{Missing, Base.SmallUnsigned} ? Union{Missing, UInt} : T

end

for i in colsidx

if eltype(ds[!, i]) >: Missing

_columns(ds)[i] = convert(Vector{Union{Missing, T}}, _columns(ds)[i])

else

_columns(ds)[i] = convert(Vector{T}, _columns(ds)[i])

end

end

init0 = _missings(T, nrow(ds))

if threads

cz = div(length(init0), Threads.nthreads())

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

if missings == :ignore

mapreduce(identity, (x,y) -> _op_for_cumsum_ignore!(x, y, lo, hi), view(_columns(ds),colsidx), init = init0)

else

mapreduce(identity, (x,y) -> _op_for_cumsum_skip!(x, y, lo, hi), view(_columns(ds),colsidx), init = init0)

end

end

else

if missings == :ignore

mapreduce(identity, (x,y) -> _op_for_cumsum_ignore!(x, y, 1, length(init0)), view(_columns(ds),colsidx), init = init0)

else

mapreduce(identity, (x,y) -> _op_for_cumsum_skip!(x, y, 1, length(init0)), view(_columns(ds),colsidx), init = init0)

end

end

removeformat!(ds, cols)

any(index(ds).sortedcols .∈ Ref(colsidx)) && _reset_grouping_info!(ds)

_modified(_attributes(ds))

ds

end

function row_cumsum(ds::AbstractDataset, cols = names(ds, Union{Missing, Number}); missings = :ignore, threads = true)

dscopy = copy(ds)

row_cumsum!(dscopy, cols, missings = missings, threads = threads)

dscopy

end

Base.@propagate_inbounds function _op_for_cumprod_skip!(x, y, lo, hi)

@simd for i in lo:hi

x[i] = _mul_prod(x[i], y[i])

end

@simd for i in lo:hi

y[i] = ifelse(ismissing(y[i]), missing, x[i])

end

x

end

Base.@propagate_inbounds function _op_for_cumprod_ignore!(x, y, lo, hi)

@simd for i in lo:hi

y[i] = _mul_prod(x[i], y[i])

end

y

end

function row_cumprod!(ds::Dataset, cols = names(ds, Union{Missing, Number}); missings = :ignore, threads = true)

!(missings in (:ignore, :skip)) && throw(ArgumentError("`missings` can be either `:ignore` or `:skip`"))

colsidx = index(ds)[cols]

T = mapreduce(eltype, promote_type, view(_columns(ds),colsidx))

for i in colsidx

if eltype(ds[!, i]) >: Missing

_columns(ds)[i] = convert(Vector{Union{Missing, T}}, _columns(ds)[i])

else

_columns(ds)[i] = convert(Vector{T}, _columns(ds)[i])

end

end

init0 = _missings(T, nrow(ds))

if threads

cz = div(length(init0), Threads.nthreads())

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

if missings == :ignore

mapreduce(identity, (x,y) -> _op_for_cumprod_ignore!(x, y, lo, hi), view(_columns(ds),colsidx), init = init0)

else

mapreduce(identity, (x,y) -> _op_for_cumprod_skip!(x, y, lo, hi), view(_columns(ds),colsidx), init = init0)

end

end

else

if missings == :ignore

mapreduce(identity, (x,y) -> _op_for_cumprod_ignore!(x, y, 1, length(init0)), view(_columns(ds),colsidx), init = init0)

else

mapreduce(identity, (x,y) -> _op_for_cumprod_skip!(x, y, 1, length(init0)), view(_columns(ds),colsidx), init = init0)

end

end

removeformat!(ds, cols)

any(index(ds).sortedcols .∈ Ref(colsidx)) && _reset_grouping_info!(ds)

_modified(_attributes(ds))

ds

end

function row_cumprod(ds::AbstractDataset, cols = names(ds, Union{Missing, Number}); missings = :ignore, threads = true)

dscopy = copy(ds)

row_cumprod!(dscopy, cols; missings = missings, threads = threads)

dscopy

end

Base.@propagate_inbounds function _op_for_cummin_skip!(x, y, lo, hi)

@simd for i in lo:hi

x[i] = _min_fun(x[i], y[i])

end

@simd for i in lo:hi

y[i] = ifelse(ismissing(y[i]), missing, x[i])

end

x

end

Base.@propagate_inbounds function _op_for_cummin_ignore!(x, y, lo, hi)

@simd for i in lo:hi

y[i] = _min_fun(x[i], y[i])

end

y

end

function row_cummin!(ds::Dataset, cols = names(ds, Union{Missing, Number}); missings = :ignore, threads = true)

!(missings in (:ignore, :skip)) && throw(ArgumentError("`missings` can be either `:ignore` or `:skip`"))

colsidx = index(ds)[cols]

T = mapreduce(eltype, promote_type, view(_columns(ds),colsidx))

for i in colsidx

if eltype(ds[!, i]) >: Missing

_columns(ds)[i] = convert(Vector{Union{Missing, T}}, _columns(ds)[i])

else

_columns(ds)[i] = convert(Vector{T}, _columns(ds)[i])

end

end

init0 = _missings(T, nrow(ds))

if threads

cz = div(length(init0), Threads.nthreads())

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

if missings == :ignore

mapreduce(identity, (x,y) -> _op_for_cummin_ignore!(x, y, lo, hi), view(_columns(ds),colsidx), init = init0)

else

mapreduce(identity, (x,y) -> _op_for_cummin_skip!(x, y, lo, hi), view(_columns(ds),colsidx), init = init0)

end

end

else

if missings == :ignore

mapreduce(identity, (x,y) -> _op_for_cummin_ignore!(x, y, 1, length(init0)), view(_columns(ds),colsidx), init = init0)

else

mapreduce(identity, (x,y) -> _op_for_cummin_skip!(x, y, 1, length(init0)), view(_columns(ds),colsidx), init = init0)

end

end

removeformat!(ds, cols)

any(index(ds).sortedcols .∈ Ref(colsidx)) && _reset_grouping_info!(ds)

_modified(_attributes(ds))

ds

end

# row_cumsum!(ds::AbstractDataset, cols = names(ds, Union{Missing, Number})) = row_cumsum!(identity, ds, cols)

function row_cummin(ds::AbstractDataset, cols = names(ds, Union{Missing, Number}); missings = :ignore, threads = true)

dscopy = copy(ds)

row_cummin!(dscopy, cols, missings = missings, threads = threads)

dscopy

end

Base.@propagate_inbounds function _op_for_cummax_skip!(x, y, lo, hi)

@simd for i in lo:hi

x[i] = _max_fun(x[i], y[i])

end

@simd for i in lo:hi

y[i] = ifelse(ismissing(y[i]), missing, x[i])

end

x

end

Base.@propagate_inbounds function _op_for_cummax_ignore!(x, y, lo, hi)

@simd for i in lo:hi

y[i] = _max_fun(x[i], y[i])

end

y

end

function row_cummax!(ds::Dataset, cols = names(ds, Union{Missing, Number}); missings = :ignore, threads = true)

!(missings in (:ignore, :skip)) && throw(ArgumentError("`missings` can be either `:ignore` or `:skip`"))

colsidx = index(ds)[cols]

T = mapreduce(eltype, promote_type, view(_columns(ds),colsidx))

for i in colsidx

if eltype(ds[!, i]) >: Missing

_columns(ds)[i] = convert(Vector{Union{Missing, T}}, _columns(ds)[i])

else

_columns(ds)[i] = convert(Vector{T}, _columns(ds)[i])

end

end

init0 = _missings(T, nrow(ds))

if threads

cz = div(length(init0), Threads.nthreads())

Threads.@threads for i in 1:Threads.nthreads()

lo = (i-1)*cz+1

i == Threads.nthreads() ? hi = length(init0) : hi = i*cz

if missings == :ignore

mapreduce(identity, (x,y) -> _op_for_cummax_ignore!(x, y, lo, hi), view(_columns(ds),colsidx), init = init0)

else

mapreduce(identity, (x,y) -> _op_for_cummax_skip!(x, y, lo, hi), view(_columns(ds),colsidx), init = init0)

end

end

else

if missings == :ignore

mapreduce(identity, (x,y) -> _op_for_cummax_ignore!(x, y, 1, length(init0)), view(_columns(ds),colsidx), init = init0)

else

mapreduce(identity, (x,y) -> _op_for_cummax_skip!(x, y, 1, length(init0)), view(_columns(ds),colsidx), init = init0)

end

end

removeformat!(ds, cols)

any(index(ds).sortedcols .∈ Ref(colsidx)) && _reset_grouping_info!(ds)

_modified(_attributes(ds))

ds

end

# row_cumsum!(ds::AbstractDataset, cols = names(ds, Union{Missing, Number})) = row_cumsum!(identity, ds, cols)

function row_cummax(ds::AbstractDataset, cols = names(ds, Union{Missing, Number}); missings = :ignore, threads = true)

dscopy = copy(ds)

row_cummax!(dscopy, cols, missings = missings, threads = threads)

dscopy

end

function row_stdze!(ds::Dataset , cols = names(ds, Union{Missing, Number}); threads = true)

colsidx = index(ds)[cols]

meandata = row_mean(ds, colsidx; threads = threads)

stddata = row_std(ds, colsidx; threads = threads)

_stdze_fun(x) = ifelse.(isequal.(stddata, 0), missing, (x .- meandata) ./ stddata)

for i in 1:length(colsidx)

_columns(ds)[colsidx[i]] = _stdze_fun(_columns(ds)[colsidx[i]])

end

removeformat!(ds, colsidx)

any(index(ds).sortedcols .∈ Ref(colsidx)) && _reset_grouping_info!(ds)

_modified(_attributes(ds))

ds

end

function row_stdze(ds::AbstractDataset , cols = names(ds, Union{Missing, Number}); threads = true)

dscopy = copy(ds)