WIP: Fix IterableTables integration, support both missing and DataValue (#204)

- add `missingtype = Missing` keyword argument where needed
- add `convertmissing` function

Author: joshday

NEWS.md

@@ -18,3 +18,8 @@
 
 - **(breaking)** Switch from DataValues to Missing.  Related: `dropna` has been changed to `dropmissing`.
 - **(breaking)** Depend on OnlineStatsBase rather than OnlineStats. 
+
+## v0.10.0
+
+- **(breaking)** Support for both DataValues and Missing (default).  When `join` generates missing values, use the keyword argument `missingtype` to set the type (`Missing` or `DataValue`)
+- Use `IndexedTables.convertmissing(tbl, T)` to convert the missing values in `tbl` to be of type `T`.

REQUIRE

@@ -2,8 +2,8 @@ julia 0.7
 OnlineStatsBase 0.9.1
 PooledArrays 0.4.1
 WeakRefStrings 0.4.4
-TableTraits 0.3.0
-TableTraitsUtils 0.2.0
 IteratorInterfaceExtensions 0.1.0
+TableTraits
 Tables
 StructArrays 0.2.0
+DataValues

src/IndexedTables.jl

@@ -1,10 +1,15 @@
 module IndexedTables
 
-using PooledArrays, SparseArrays, Statistics, WeakRefStrings, TableTraits, 
-    TableTraitsUtils, IteratorInterfaceExtensions
+using PooledArrays, SparseArrays, Statistics, WeakRefStrings
 
 using OnlineStatsBase: OnlineStat, fit!
-import Tables
+using StructArrays: StructVector, StructArray, foreachfield, fieldarrays, 
+    collect_structarray, staticschema, ArrayInitializer, refine_perm!, collect_structarray, 
+    collect_empty_structarray, grow_to_structarray!, collect_to_structarray! 
+
+import Tables, TableTraits, IteratorInterfaceExtensions
+
+import DataValues: DataValue, DataValueArray, isna
 
 import Base:
     show, eltype, length, getindex, setindex!, ndims, map, convert, keys, values,
@@ -14,9 +19,6 @@ import Base:
     tuple_type_cons, tuple_type_head, tuple_type_tail, in, convert
 
 
-using StructArrays: StructVector, StructArray, foreachfield, fieldarrays, collect_structarray, staticschema, ArrayInitializer,
-                    refine_perm!, collect_structarray, collect_empty_structarray, grow_to_structarray!, collect_to_structarray! 
-
 #-----------------------------------------------------------------------# exports
 export 
     # macros
@@ -77,9 +79,7 @@ include("reduce.jl")
 include("flatten.jl")
 include("join.jl")
 include("reshape.jl")
-
-# TableTraits/Tables integration
-include("tabletraits.jl")
 include("tables.jl")
+include("tabletraits.jl")
 
 end # module

src/collect.jl

@@ -1,4 +1,4 @@
-const default_initializer = ArrayInitializer(t -> t<:Union{Tuple, NamedTuple, Pair})
+const default_initializer = ArrayInitializer(t -> t<:Union{Tuple, NamedTuple, Pair}, (T, sz) -> similar(arrayof(T), sz))
 
 """
     collect_columns(itr)

src/columns.jl

@@ -423,6 +423,9 @@ function ColDict(t; copy=nothing)
     ColDict(Int[], t, convert(Array{Any}, collect(cnames)), Any[columns(t)...], copy)
 end
 
+Base.keys(d::ColDict) = d.names
+Base.values(d::ColDict) = d.columns
+
 function Base.getindex(d::ColDict{<:Columns})
     Columns(Tuple(d.columns); names=d.names)
 end

src/indexedtable.jl

@@ -131,12 +131,9 @@ _impl(impl::Val) = impl
 _impl(impl::Val, x::AbstractArray, z...) = _impl(impl, z...)
 _impl(x::AbstractArray...) = _impl(Val{:serial}(), x...)
 
+# table methods go through here
 function table(cs::Tup; chunks=nothing, kwargs...)
-    if chunks !== nothing
-        impl = Val{:distributed}()
-    else
-        impl = _impl(astuple(cs)...)
-    end
+    impl = chunks !== nothing ? Val(:distributed) : _impl(astuple(cs)...)
     table(impl, cs; chunks=chunks, kwargs...)
 end
 

src/join.jl

@@ -50,7 +50,7 @@ end
 end
 
 function _join!(::Val{typ}, ::Val{grp}, ::Val{keepkeys}, f, I, data, ks, lout, rout,
-      lnull, rnull, lkey, rkey, ldata, rdata, lperm, rperm, init_group, accumulate) where {typ, grp, keepkeys}
+      lnull, rnull, lkey, rkey, ldata, rdata, lperm, rperm, init_group, accumulate, missingtype) where {typ, grp, keepkeys}
 
     ll, rr = length(lkey), length(rkey)
 
@@ -102,7 +102,8 @@ function _join!(::Val{typ}, ::Val{grp}, ::Val{keepkeys}, f, I, data, ks, lout, r
                             # optimized push! method for concat_tup
                             _push!(Val{:both}(), f, data,
                                    lout, rout, ldata, rdata,
-                                   lperm[x], rperm[y], missing, missing)
+                                   lperm[x], rperm[y], 
+                                   missing_instance(missingtype), missing_instance(missingtype))
                         end
                     end
                 else
@@ -167,11 +168,36 @@ function _join!(::Val{typ}, ::Val{grp}, ::Val{keepkeys}, f, I, data, ks, lout, r
     lnull_idx, rnull_idx
 end
 
-nullrow(t::Type{<:Tuple}) = Tuple(map(x->missing, fieldtypes(t)))
-nullrow(t::Type{<:NamedTuple}) = t(Tuple(map(x->missing, fieldtypes(t))))
-nullrow(t) = missing
 
-function init_join_output(typ, grp, f, ldata, rdata, left, keepkeys, lkey, rkey, init_group, accumulate)
+# Missing
+nullrow(t::Type{<:Tuple}, ::Type{Missing}) = Tuple(map(x->missing, fieldtypes(t)))
+nullrow(t::Type{<:NamedTuple}, ::Type{Missing}) = t(Tuple(map(x->missing, fieldtypes(t))))
+nullrow(t, ::Type{Missing}) = missing
+function outvec(col, idxs, ::Type{Missing})
+    v = convert(Vector{Union{Missing, eltype(col)}}, col)
+    v[idxs] .= missing
+    v
+end
+
+
+# DataValue
+nullrow(::Type{T}, ::Type{DataValue}) where {T <: Tuple} = Tuple(fieldtype(T, i)() for i = 1:fieldcount(T))
+function nullrow(::Type{NamedTuple{names, T}}, ::Type{DataValue}) where {names, T} 
+    NamedTuple{names, T}(Tuple(fieldtype(T, i)() for i = 1:fieldcount(T)))
+end
+nullrow(t, ::Type{DataValue}) = DataValue()
+nullrow(t::DataValue, ::Type{DataValue}) = t()
+function outvec(col, idxs, ::Type{DataValue})
+    nulls = zeros(Bool, length(col))
+    nulls[idxs] .= true
+    if col isa DataValueArray
+        col.isna[idxs] .= true
+    else
+        DataValueArray(col, nulls)
+    end
+end
+
+function init_join_output(typ, grp, f, ldata, rdata, left, keepkeys, lkey, rkey, init_group, accumulate, missingtype)
     lnull = nothing
     rnull = nothing
     loutput = nothing
@@ -181,16 +207,16 @@ function init_join_output(typ, grp, f, ldata, rdata, left, keepkeys, lkey, rkey,
 
         left_type = eltype(ldata)
         if !isa(typ, Union{Val{:left}, Val{:inner}, Val{:anti}})
-            null_left_type = map_params(x -> Union{Missing, x}, eltype(ldata))
-            lnull = nullrow(null_left_type)
+            null_left_type = map_params(x -> type2missingtype(x, missingtype), eltype(ldata))
+            lnull = nullrow(null_left_type, missingtype)
         else
             null_left_type = left_type
         end
 
         right_type = eltype(rdata)
         if !isa(typ, Val{:inner})
-            null_right_type = map_params(x->Union{Missing, x}, eltype(rdata))
-            rnull = nullrow(null_right_type)
+            null_right_type = map_params(x -> type2missingtype(x, missingtype), eltype(rdata))
+            rnull = nullrow(null_right_type, missingtype)
         else
             null_right_type = right_type
         end
@@ -253,11 +279,19 @@ with each `right` row, resulting in `n_occurrences_left * n_occurrences_right` o
 
 # Options (keyword arguments)
 
-- `how = :inner` -- join method to use. Described below.
-- `lkey = pkeys(left)` -- fields from `left` to match on (see [`pkeys`](@ref))
-- `rkey = pkeys(right)` -- fields from `right` to match on
-- `lselect = Not(lkey)` -- output columns from `left` (see [`Not`](@ref))
-- `rselect = Not(rkey)` -- output columns from `right`
+- `how = :inner` 
+    - Join method to use. Described below.
+- `lkey = pkeys(left)` 
+    - Fields from `left` to match on (see [`pkeys`](@ref)).
+- `rkey = pkeys(right)` 
+    - Fields from `right` to match on.
+- `lselect = Not(lkey)` 
+    - Output columns from `left` (see [`Not`](@ref))
+- `rselect = Not(rkey)`
+    - Output columns from `right`.
+- `missingtype = Missing` 
+    - Type of missing values that can be created through `:left` and `:outer` joins.
+    - Other supported option is `DataValue`.
 
 ## Join methods (`how = :inner`)
 
@@ -289,7 +323,8 @@ function Base.join(f, left::Dataset, right::Dataset;
                    keepkeys=false, # defaults to keeping the keys for only the joined columns
                    init_group=nothing,
                    accumulate=nothing,
-                   cache=true)
+                   cache=true,
+                   missingtype=Missing)
 
     if !(how in [:inner, :left, :outer, :anti])
         error("Invalid how: supported join types are :inner, :left, :outer, and :anti")
@@ -331,26 +366,18 @@ function Base.join(f, left::Dataset, right::Dataset;
     I, data, ks, lout, rout, lnull, rnull, init_group, accumulate =
         init_join_output(typ, grp, f, ldata, rdata,
                          left, keepkeys, lkey, rkey,
-                         init_group, accumulate)
+                         init_group, accumulate, missingtype)
 
     lnull_idx, rnull_idx = _join!(typ, grp, Val{keepkeys}(), f, I,
                                   data, ks, lout, rout, lnull, rnull,
                                   lkey, rkey, ldata, rdata, lperm,
-                                  rperm, init_group, accumulate)
+                                  rperm, init_group, accumulate, missingtype)
 
     if !isempty(lnull_idx) && lout !== nothing
-        lnulls = zeros(Bool, length(lout))
-        lnulls[lnull_idx] .= true
         lout = if lout isa Columns
-            Columns(map(columns(lout)) do col
-                v = convert(Vector{Union{Missing, eltype(col)}}, col)
-                v[lnull_idx] .= missing
-                v
-            end)
+            Columns(map(col -> outvec(col, lnull_idx, missingtype), columns(lout)))
         else
-            v = convert(Vector{Union{Missing, eltype(lout)}}, lout)
-            v[lnull_idx] .= missing
-            v
+            outvec(col, lnull_idx, missingtype)
         end
         data = concat_cols(lout, rout)
     end
@@ -359,15 +386,9 @@ function Base.join(f, left::Dataset, right::Dataset;
         rnulls = zeros(Bool, length(rout))
         rnulls[rnull_idx] .= true
         rout = if rout isa Columns
-            Columns(map(columns(rout)) do col
-                v = convert(Vector{Union{Missing, eltype(col)}}, col)
-                v[rnull_idx] .= missing
-                v
-            end)
+            Columns(map(col -> outvec(col, rnull_idx, missingtype), columns(rout)))
         else
-            v = convert(Vector{Union{Missing, eltype(rout)}}, rout)
-            v[rnull_idx] .= missing
-            v
+            outvec(col, rnull_idx, missingtype)
         end
         data = concat_cols(lout, rout)
     end

src/selection.jl

@@ -11,6 +11,7 @@ Select all or a subset of columns, or a single column from the table.
 4. `AbstractArray` -- returns the array itself. This must be the same length as the table.
 5. `Tuple` of `Selection` -- returns a table containing a column for every selector in the tuple. The tuple may also contain the type `Pair{Symbol, Selection}`, which the selection a name. The most useful form of this when introducing a new column.
 6. `Regex` -- returns the columns with names that match the regular expression.
+7. `Type` -- returns columns with elements of the given type.
 
 # Examples:
 
@@ -122,7 +123,7 @@ Returns a new table if `f` returns a tuple or named tuple.  If not, returns a ve
 function map(f, t::AbstractIndexedTable; select=nothing) end
 
 function map(f, t::Dataset; select=nothing, copy=false, kwargs...)
-    if isa(f, Tup) && select===nothing
+    if isa(f, Tup) && select === nothing
         select = colnames(t)
     elseif select === nothing
         select = valuenames(t)
@@ -132,48 +133,90 @@ function map(f, t::Dataset; select=nothing, copy=false, kwargs...)
     isa(x, Columns) ? table(x; copy=false, kwargs...) : x
 end
 
-function _non_missing(t::Union{Columns, IndexedTable}, sel=(colnames(t)...,))
+
+missing_indxs(v::Vector) = findall(!_ismissing, v)
+
+function missing_indxs(t::StructArray)
     indxs = collect(1:length(t))
-    by = isa(sel, Tuple) ? sel : (sel,)
-    bycols = columns(t, by)
-    d = ColDict(t)
-    for (key, c) in zip(by, bycols)
-        x = rows(t, c)
-        if Missing <: eltype(x)
-            filt_by_col!(!ismissing, x, indxs)
-            y = Vector{Base.nonmissingtype(eltype(x))}(undef, length(x))
-            y[indxs] = x[indxs]
-            d[key] = y
-        else
-            d[key] = x
-        end
+    for vec in getfield(t, :fieldarrays)
+        filter!(i -> !_ismissing(vec[i]), indxs)
     end
-    (d[], indxs)
+    indxs
 end
 
 """
-    dropmissing(t)
+    dropmissing(t        )
     dropmissing(t, select)
 
-Drop rows of table `t` which contain `missing` values, optionally only 
-using the columns in `select`.  
+Drop rows of table `t` which contain missing values (either `Missing` or `DataValue`), 
+optionally only using the columns in `select`.  Column types will be converted to 
+non-missing types.  For example:
 
-Column types will be converted to non-`Missing` types.  E.g. `Array{Union{Int, Missing}}` 
-to `Array{Int}`.
+- `Vector{Union{Int, Missing}}` -> `Vector{Int}`
+- `DataValueArray{Int}` -> Vector{Int}
 
 # Example
 
     t = table([0.1,0.5,missing,0.7], [2,missing,4,5], [missing,6,missing,7], names=[:t,:x,:y])
     dropmissing(t)
     dropmissing(t, (:t, :x))
 """
-function dropmissing(t::Dataset, by=colnames(t))
-    subtable(_non_missing(t, by)...)
+function dropmissing(t::IndexedTable, sel = All())
+    selection = lowerselection(t, sel)
+    indxs = missing_indxs(rows(t, selection))
+    t2 = subtable(t, indxs)
+    d = ColDict(t2)
+    for s in selection
+        T = eltype(d[s])
+        d[s] = convert(Vector{missingtype2type(T)}, d[s])
+    end
+    
+    table(d[], copy=false, perms=t.perms, presorted=true)
 end
 
-@deprecate dropna dropmissing
+dropmissing(t::NDSparse, sel=All()) = ndsparse(dropmissing(table(t), sel))
+
+Base.@deprecate_binding dropna dropmissing
+
+"""
+    convertmissing(tbl, missingtype)
+
+Convert the missing value representation in `tbl` to be of type `missingtype`.
+
+# Example
+
+    using IndexedTables, DataValues
+    t = table([1,2,missing], [1,missing,3])
+    IndexedTables.convertmissing(t, DataValue)
+"""
+function convertmissing(t::IndexedTable, ::Type{Missing}) 
+    d = ColDict(t)
+    for (k, v) in pairs(d)
+        T = eltype(v)
+        if T <: DataValue
+            indxs = findall(!isna, v)
+            y = Vector{Union{Missing, missingtype2type(T)}}(missing, length(v))
+            y[indxs] = get.(v[indxs])
+            d[k] = y
+        end
+    end
+    subtable(d[], 1:length(t))
+end
+function convertmissing(t::IndexedTable, ::Type{DataValue}) 
+    d = ColDict(t)
+    for (k, v) in pairs(d)
+        T = eltype(v)
+        if Missing <: T 
+            indxs = findall(!ismissing, v)
+            y = DataValueArray(Vector{Base.nonmissingtype(T)}(undef, length(v)), trues(length(v)))
+            y[indxs] = v[indxs]
+            d[k] = y
+        end
+    end
+    subtable(d[], 1:length(t))
+end
+convertmissing(t::NDSparse, Typ) = ndsparse(convertmissing(table(t), Typ))
 
-filt_by_col!(f, col, indxs) = filter!(i->f(col[i]), indxs)
 
 """
     filter(f, t::Union{IndexedTable, NDSparse}; select)
@@ -215,6 +258,8 @@ function Base.filter(pred::Tuple, t::Dataset; select=nothing)
     subtable(t, indxs, presorted=true)
 end
 
+filt_by_col!(f, col, indxs) = filter!(i->f(col[i]), indxs)
+
 function Base.filter(pred::Pair, t::Dataset; select=nothing)
     filter((pred,), t, select=select)
 end