Add parallel reduction supports for RowIterator and NamedTupleIterator

Project.toml

@@ -10,12 +10,14 @@ TableTraits = "3783bdb8-4a98-5b6b-af9a-565f29a5fe9c"
 IteratorInterfaceExtensions = "82899510-4779-5014-852e-03e436cf321d"
 Test = "8dfed614-e22c-5e08-85e1-65c5234f0b40"
 DataValueInterfaces = "e2d170a0-9d28-54be-80f0-106bbe20a464"
+SplittablesBase = "171d559e-b47b-412a-8079-5efa626c420e"
 
 [extras]
 Test = "8dfed614-e22c-5e08-85e1-65c5234f0b40"
 DataValues = "e7dc6d0d-1eca-5fa6-8ad6-5aecde8b7ea5"
 QueryOperators = "2aef5ad7-51ca-5a8f-8e88-e75cf067b44b"
 SparseArrays = "2f01184e-e22b-5df5-ae63-d93ebab69eaf"
+SplittablesTesting = "3bda5eb5-c32a-4f64-8618-df3be8968470"
 
 [compat]
 julia = "1"
@@ -25,4 +27,4 @@ IteratorInterfaceExtensions = "0.1.1, 1"
 TableTraits = "0.4.1, 1"
 
 [targets]
-test = ["Test", "DataValues", "QueryOperators", "SparseArrays"]
+test = ["Test", "DataValues", "QueryOperators", "SparseArrays", "SplittablesTesting"]

src/Tables.jl

@@ -1,6 +1,6 @@
 module Tables
 
-using LinearAlgebra, DataValueInterfaces, DataAPI, TableTraits, IteratorInterfaceExtensions
+using LinearAlgebra, DataValueInterfaces, DataAPI, TableTraits, IteratorInterfaceExtensions, SplittablesBase
 
 export rowtable, columntable
 

src/fallbacks.jl

@@ -26,7 +26,7 @@ Base.@propagate_inbounds getcolumn(c::ColumnsRow, i::Int) = getcolumn(getcolumns
 Base.@propagate_inbounds getcolumn(c::ColumnsRow, nm::Symbol) = getcolumn(getcolumns(c), nm)[getrow(c)]
 columnnames(c::ColumnsRow) = columnnames(getcolumns(c))
 
-@generated function Base.isless(c::ColumnsRow{T}, d::ColumnsRow{T}) where {T <: NamedTuple{names}} where names
+@generated function Base.isless(c::ColumnsRow{<:NamedTuple{names}}, d::ColumnsRow{<:NamedTuple{names}}) where names
     exprs = Expr[]
     for n in names
         var1 = Expr(:., :c, QuoteNode(n))
@@ -42,7 +42,7 @@ columnnames(c::ColumnsRow) = columnnames(getcolumns(c))
     Expr(:block, exprs...)
 end
 
-@generated function Base.isequal(c::ColumnsRow{T}, d::ColumnsRow{T}) where {T <: NamedTuple{names}} where names
+@generated function Base.isequal(c::ColumnsRow{<:NamedTuple{names}}, d::ColumnsRow{<:NamedTuple{names}}) where names
     exprs = Expr[]
     for n in names
         var1 = Expr(:., :c, QuoteNode(n))
@@ -79,6 +79,28 @@ schema(x::RowIterator) = schema(columns(x))
     return ColumnsRow(columns(rows), st), st + 1
 end
 
+function consistent_rowcount(cols)
+    len = length(cols[1])
+    if !all(c -> length(c) == len, cols)
+        throw(ArgumentError("`halve` on columns return inconsistent number or rows"))
+    end
+    return len
+end
+
+function SplittablesBase.halve(x::RowIterator)
+    if isempty(columns(x))
+        len = cld(length(x), 2)
+        return (RowIterator(columns(x), len), RowIterator(columns(x), length(x) - len))
+    end
+    cs = map(SplittablesBase.halve, columns(x))
+    lefts = map(first, cs)
+    rights = map(last, cs)
+    return (
+        RowIterator(lefts, consistent_rowcount(lefts)),
+        RowIterator(rights, consistent_rowcount(rights)),
+    )
+end
+
 # this is our generic Tables.rows fallback definition
 function rows(x::T) where {T}
     isrowtable(x) && return x

src/namedtuples.jl

@@ -65,6 +65,14 @@ function Base.iterate(rows::NamedTupleIterator{Nothing}, state::Tuple{Val{names}
     return NamedTuple{names}(Tuple(getcolumn(row, nm) for nm in names)), (Val(names), (st,))
 end
 
+function SplittablesBase.halve(rows::NamedTupleIterator{schema}) where schema
+    left, right = SplittablesBase.halve(rows.x)
+    return (
+        NamedTupleIterator{schema,typeof(left)}(left),
+        NamedTupleIterator{schema,typeof(right)}(right),
+    )
+end
+
 # sink function
 """
     Tables.rowtable(x) => Vector{NamedTuple}

test/runtests.jl

@@ -1,4 +1,4 @@
-using Test, Tables, TableTraits, DataValues, QueryOperators, IteratorInterfaceExtensions, SparseArrays
+using Test, Tables, TableTraits, DataValues, QueryOperators, IteratorInterfaceExtensions, SparseArrays, SplittablesBase, SplittablesTesting
 
 @testset "utils.jl" begin
 
@@ -551,3 +551,34 @@ Tables.isrowtable(::Type{IsRowTable}) = true
     @test Tables.columntable(rt) == Tables.columntable([nt, nt, nt])
 
 end
+
+@testset "SplittablesBase" begin
+    nt4 = (a = [0, 1, 2, 3], b = [5, 6, 7, 8])
+    nt5 = (a = [0, 1, 2, 3, 4], b = [5, 6, 7, 8, 9])
+    nt0 = NamedTuple()
+    SplittablesTesting.test_ordered([
+        (label = "RowIterator (length = 4)", data = Tables.rows(nt4)),
+        (label = "RowIterator (length = 5)", data = Tables.rows(nt5)),
+        (label = "RowIterator (no columns)", data = Tables.RowIterator(nt0, 5)),
+        (
+            label = "NamedTupleIterator (length = 4)",
+            data = Tables.namedtupleiterator(Tables.rows(nt4)),
+        ),
+        (
+            label = "NamedTupleIterator (length = 5)",
+            data = Tables.namedtupleiterator(Tables.rows(nt5)),
+        ),
+        (
+            label = "NamedTupleIterator (no columns)",
+            data = Tables.namedtupleiterator(Tables.RowIterator(nt0, 5)),
+        ),
+    ])
+
+    @testset "Inconsistent `halve` of columns should throw" begin
+        rt = Tables.rows((a = [0, 1, 2, 3, 4], b = [5, 6, 7, 8]))
+        @test_throws(
+            ArgumentError("`halve` on columns return inconsistent number or rows"),
+            SplittablesBase.halve(rt)
+        )
+    end
+end