Add the LINQ operators Query.jl is missing

NEWS.md

@@ -1,3 +1,13 @@
+# QueryOperators.jl v1.2.0 Release Notes
+* Add the left_join, right_join and full_join operators
+* Add the concat, union, union_by, except, except_by, intersect and intersect_by operators
+* Add the take_while, drop_while, take_last and drop_last operators
+* Add the order, order_descending, reverse, shuffle and index operators
+* Add the append, prepend and zip operators
+* Add the count_by, aggregate_by and chunk operators
+* Add the of_type and cast operators
+* Add the min_by, max_by, any, all, contains, sequence_equal, aggregate, first, last, single and element_at terminal operators
+
 # QueryOperators.jl v1.1.0 Release Notes
 * Add the summarize operator for grouped and whole-table aggregation
 

Project.toml

@@ -1,12 +1,13 @@
 name = "QueryOperators"
 uuid = "2aef5ad7-51ca-5a8f-8e88-e75cf067b44b"
-version = "1.1.1-DEV"
+version = "1.2.0-DEV"
 
 [deps]
 DataStructures = "864edb3b-99cc-5e75-8d2d-829cb0a9cfe8"
 TableShowUtils = "5e66a065-1f0a-5976-b372-e0b8c017ca10"
 DataValues = "e7dc6d0d-1eca-5fa6-8ad6-5aecde8b7ea5"
 IteratorInterfaceExtensions = "82899510-4779-5014-852e-03e436cf321d"
+Random = "9a3f8284-a2c9-5f02-9a11-845980a1fd5c"
 
 [extras]
 TestItemRunner = "f8b46487-2199-4994-9208-9a1283c18c0a"
@@ -18,6 +19,7 @@
 TableShowUtils = "0.1.1, 0.2, 0.3, 1"
 DataValues = "0.4.4, 0.5, 1"
 IteratorInterfaceExtensions = "0.1.1, 1"
+Random = "1"
 
 [targets]
 test = ["Test", "TestItemRunner"]

src/QueryOperators.jl

@@ -4,6 +4,7 @@ using DataStructures
 using IteratorInterfaceExtensions
 using TableShowUtils
 import DataValues
+import Random
 
 export Grouping, key
 
@@ -14,6 +15,9 @@ include("enumerable/enumerable.jl")
 include("enumerable/enumerable_groupby.jl")
 include("enumerable/enumerable_join.jl")
 include("enumerable/enumerable_groupjoin.jl")
+include("enumerable/enumerable_leftjoin.jl")
+include("enumerable/enumerable_rightjoin.jl")
+include("enumerable/enumerable_fulljoin.jl")
 include("enumerable/enumerable_orderby.jl")
 include("enumerable/enumerable_map.jl")
 include("enumerable/enumerable_filter.jl")
@@ -22,9 +26,28 @@ include("enumerable/enumerable_defaultifempty.jl")
 include("enumerable/enumerable_count.jl")
 include("enumerable/enumerable_take.jl")
 include("enumerable/enumerable_drop.jl")
+include("enumerable/enumerable_takewhile.jl")
+include("enumerable/enumerable_dropwhile.jl")
+include("enumerable/enumerable_takelast.jl")
+include("enumerable/enumerable_droplast.jl")
 include("enumerable/enumerable_unique.jl")
+include("enumerable/enumerable_concat.jl")
+include("enumerable/enumerable_union.jl")
+include("enumerable/enumerable_except.jl")
+include("enumerable/enumerable_intersect.jl")
+include("enumerable/enumerable_reverse.jl")
+include("enumerable/enumerable_shuffle.jl")
+include("enumerable/enumerable_index.jl")
+include("enumerable/enumerable_append.jl")
+include("enumerable/enumerable_prepend.jl")
+include("enumerable/enumerable_zip.jl")
+include("enumerable/enumerable_oftype.jl")
 include("enumerable/enumerable_pivot.jl")
 include("enumerable/enumerable_summarize.jl")
+include("enumerable/enumerable_countby.jl")
+include("enumerable/enumerable_aggregateby.jl")
+include("enumerable/enumerable_chunk.jl")
+include("enumerable/enumerable_terminal.jl")
 include("enumerable/show.jl")
 
 include("source_iterable.jl")

src/enumerable/enumerable.jl

@@ -1,6 +1,29 @@
 abstract type Enumerable end
 
 Base.IteratorSize(::Type{T}) where {T <: Enumerable} = Base.SizeUnknown()
 IteratorInterfaceExtensions.isiterable(x::Enumerable) = true
 
 haslength(S) = Base.IteratorSize(S) isa Union{Base.HasLength, Base.HasShape} ? Base.HasLength() : Base.IteratorSize(S)
+
+# Operators that walk more than one source keep "which source, how far into it"
+# in their iteration state. `_NotStarted` marks a source that has not been
+# iterated yet, so that `_iterate_from` can pick the right `iterate` method
+# without conflating it with a source whose own state happens to be `nothing`.
+struct _NotStarted end
+
+_iterate_from(source, ::_NotStarted) = iterate(source)
+_iterate_from(source, state) = iterate(source, state)
+
+# Element types of two sources that will be emitted into a single stream have
+# to agree, the same requirement `default_if_empty` places on its default value.
+function _check_same_eltype(op, ::Type{T1}, ::Type{T2}) where {T1,T2}
+    if T1 != T2
+        error("The two sequences passed to $op have different element types, $T1 and $T2.")
+    end
+end
+
+function _check_same_keytype(op, ::Type{TKey1}, ::Type{TKey2}) where {TKey1,TKey2}
+    if TKey1 != TKey2
+        error("The keys of the two sequences passed to $op have different types, $TKey1 and $TKey2.")
+    end
+end

src/enumerable/enumerable_aggregateby.jl

@@ -0,0 +1,49 @@
+_aggregate_by_row(k, v) = Base.merge(_key_namedtuple(k), (value = v,))
+
+struct EnumerableAggregateBy{T,TKey,TACC,S,Q<:Function,A<:Function} <: Enumerable
+    source::S
+    f::Q
+    seed::TACC
+    accumulator::A
+end
+
+Base.eltype(::Type{EnumerableAggregateBy{T,TKey,TACC,S,Q,A}}) where {T,TKey,TACC,S,Q,A} = T
+
+# Enumerable.AggregateBy (.NET 9): fold the elements of each key into a single
+# value, without materialising the intermediate groupings. `accumulator` is
+# called as `accumulator(accumulated, element)`, matching .NET's argument order.
+#
+# `summarize` is the more general and more idiomatic way to aggregate here;
+# `aggregate_by` exists for LINQ parity.
+function aggregate_by(source::Enumerable, f::Function, f_expr::Expr, seed, accumulator::Function)
+    TS = eltype(source)
+    TKey = Base._return_type(f, Tuple{TS,})
+    TACC = typeof(seed)
+    T = Base._return_type(_aggregate_by_row, Tuple{TKey,TACC})
+
+    return EnumerableAggregateBy{T,TKey,TACC,typeof(source),typeof(f),typeof(accumulator)}(source, f, seed, accumulator)
+end
+
+function Base.iterate(iter::EnumerableAggregateBy{T,TKey,TACC,S,Q,A}) where {T,TKey,TACC,S,Q,A}
+    accumulated = OrderedDict{TKey,TACC}()
+    for i in iter.source
+        k = iter.f(i)
+        accumulated[k] = iter.accumulator(get(accumulated, k, iter.seed), i)
+    end
+
+    rows = T[_aggregate_by_row(k, v) for (k, v) in accumulated]
+
+    if length(rows)==0
+        return nothing
+    end
+
+    return rows[1], (rows, 2)
+end
+
+function Base.iterate(iter::EnumerableAggregateBy{T,TKey,TACC,S,Q,A}, state) where {T,TKey,TACC,S,Q,A}
+    if state[2]>length(state[1])
+        return nothing
+    else
+        return state[1][state[2]], (state[1], state[2]+1)
+    end
+end

src/enumerable/enumerable_append.jl

@@ -0,0 +1,32 @@
+struct EnumerableAppend{T,S} <: Enumerable
+    source::S
+    element::T
+end
+
+# Enumerable.Append: the source followed by one more element. The element is
+# converted to the source's element type, so appending an Int to a sequence of
+# Float64 works.
+function append(source::Enumerable, element)
+    T = eltype(source)
+    return EnumerableAppend{T,typeof(source)}(source, convert(T, element))
+end
+
+Base.IteratorSize(::Type{EnumerableAppend{T,S}}) where {T,S} = haslength(S)
+
+Base.eltype(::Type{EnumerableAppend{T,S}}) where {T,S} = T
+
+Base.length(iter::EnumerableAppend) = length(iter.source) + 1
+
+Base.iterate(iter::EnumerableAppend) = _append_next(iter, _NotStarted())
+
+function Base.iterate(iter::EnumerableAppend, state)
+    # `state.done` marks the appended element as already handed out.
+    state.done && return nothing
+    return _append_next(iter, state.state)
+end
+
+function _append_next(iter::EnumerableAppend, source_state)
+    ret = _iterate_from(iter.source, source_state)
+    ret === nothing && return iter.element, (done=true, state=source_state)
+    return ret[1], (done=false, state=ret[2])
+end

src/enumerable/enumerable_chunk.jl

@@ -0,0 +1,42 @@
+struct EnumerableChunk{T,TI,S} <: Enumerable
+    source::S
+    n::Int
+end
+
+# Enumerable.Chunk (.NET 6): split the source into batches of at most `n`
+# elements. The final batch is shorter when the source does not divide evenly.
+function chunk(source::Enumerable, n::Integer)
+    n < 1 && error("The chunk size must be at least 1, got $n.")
+
+    TI = eltype(source)
+    T = Vector{TI}
+
+    return EnumerableChunk{T,TI,typeof(source)}(source, Int(n))
+end
+
+Base.IteratorSize(::Type{EnumerableChunk{T,TI,S}}) where {T,TI,S} = haslength(S)
+
+Base.eltype(::Type{EnumerableChunk{T,TI,S}}) where {T,TI,S} = T
+
+Base.length(iter::EnumerableChunk) = cld(length(iter.source), iter.n)
+
+Base.iterate(iter::EnumerableChunk) = _chunk_next(iter, _NotStarted())
+
+Base.iterate(iter::EnumerableChunk, state) = _chunk_next(iter, state)
+
+# Pulls at most `n` elements per call, so a chunked source is only walked as
+# far as the batches actually consumed.
+function _chunk_next(iter::EnumerableChunk{T,TI,S}, source_state) where {T,TI,S}
+    buffer = TI[]
+
+    while length(buffer) < iter.n
+        ret = _iterate_from(iter.source, source_state)
+        ret === nothing && break
+        push!(buffer, ret[1])
+        source_state = ret[2]
+    end
+
+    length(buffer)==0 && return nothing
+
+    return buffer, source_state
+end

src/enumerable/enumerable_concat.jl

@@ -0,0 +1,44 @@
+struct EnumerableConcat{T,S1,S2} <: Enumerable
+    first::S1
+    second::S2
+end
+
+function concat(first::Enumerable, second::Enumerable)
+    T1 = eltype(first)
+    T2 = eltype(second)
+
+    _check_same_eltype("concat", T1, T2)
+
+    return EnumerableConcat{T1,typeof(first),typeof(second)}(first, second)
+end
+
+Base.eltype(::Type{EnumerableConcat{T,S1,S2}}) where {T,S1,S2} = T
+
+function Base.IteratorSize(::Type{EnumerableConcat{T,S1,S2}}) where {T,S1,S2}
+    return haslength(S1) isa Base.HasLength && haslength(S2) isa Base.HasLength ?
+        Base.HasLength() : Base.SizeUnknown()
+end
+
+Base.length(iter::EnumerableConcat) = length(iter.first) + length(iter.second)
+
+Base.iterate(iter::EnumerableConcat) = _concat_next(iter, 1, _NotStarted())
+
+function Base.iterate(iter::EnumerableConcat, state)
+    return _concat_next(iter, state.side, state.state)
+end
+
+function _concat_next(iter::EnumerableConcat, side, source_state)
+    if side == 1
+        ret = _iterate_from(iter.first, source_state)
+        if ret !== nothing
+            return ret[1], (side=1, state=ret[2])
+        end
+        # First source exhausted — fall through to the second.
+        side = 2
+        source_state = _NotStarted()
+    end
+
+    ret = _iterate_from(iter.second, source_state)
+    ret === nothing && return nothing
+    return ret[1], (side=2, state=ret[2])
+end

src/enumerable/enumerable_countby.jl

@@ -0,0 +1,44 @@
+# Key columns are named exactly as `summarize` names them: a scalar key becomes
+# a column called `key`, a NamedTuple key contributes one column per field.
+_count_by_row(k, n::Int) = Base.merge(_key_namedtuple(k), (count = n,))
+
+struct EnumerableCountBy{T,TKey,S,Q<:Function} <: Enumerable
+    source::S
+    f::Q
+end
+
+Base.eltype(::Type{EnumerableCountBy{T,TKey,S,Q}}) where {T,TKey,S,Q} = T
+
+# Enumerable.CountBy (.NET 9): the frequency of each key, without materialising
+# the intermediate groupings that `groupby` would build.
+function count_by(source::Enumerable, f::Function, f_expr::Expr)
+    TS = eltype(source)
+    TKey = Base._return_type(f, Tuple{TS,})
+    T = Base._return_type(_count_by_row, Tuple{TKey,Int})
+
+    return EnumerableCountBy{T,TKey,typeof(source),typeof(f)}(source, f)
+end
+
+function Base.iterate(iter::EnumerableCountBy{T,TKey,S,Q}) where {T,TKey,S,Q}
+    counts = OrderedDict{TKey,Int}()
+    for i in iter.source
+        k = iter.f(i)
+        counts[k] = get(counts, k, 0) + 1
+    end
+
+    rows = T[_count_by_row(k, n) for (k, n) in counts]
+
+    if length(rows)==0
+        return nothing
+    end
+
+    return rows[1], (rows, 2)
+end
+
+function Base.iterate(iter::EnumerableCountBy{T,TKey,S,Q}, state) where {T,TKey,S,Q}
+    if state[2]>length(state[1])
+        return nothing
+    else
+        return state[1][state[2]], (state[1], state[2]+1)
+    end
+end

src/enumerable/enumerable_defaultifempty.jl

@@ -3,7 +3,7 @@
     default_value::T
 end
 
 Base.eltype(iter::Type{EnumerableDefaultIfEmpty{T,S}}) where {T,S} = T
 
 _default_value_expr(::Type{T}) where {T} = :( DataValues.DataValue{$T}() )
 
@@ -13,6 +13,14 @@
     return :( NamedTuple{$(fieldnames(T))}( ($( (_default_value_expr(fieldtype(T,i)) for i in 1:length(fieldnames(T)))...   ),)) )
 end
 
+# Runtime counterpart of `_default_value_expr`, used by the outer join operators
+# to build the all-null element that an unmatched side contributes. Generated so
+# that the NamedTuple case is constructed at compile time, exactly as
+# `default_if_empty` below does.
+@generated function _default_value(::Type{T}) where {T}
+    return _default_value_expr(T)
+end
+
 @generated function default_if_empty(source::S) where {S}
     T_source = eltype(source)
 

src/enumerable/enumerable_droplast.jl

@@ -0,0 +1,48 @@
+struct EnumerableDropLast{T,S} <: Enumerable
+    source::S
+    n::Int
+end
+
+# Enumerable.SkipLast: everything but the trailing `n` elements. A count of
+# zero or less leaves the source unchanged, as in .NET.
+function drop_last(source::Enumerable, n::Integer)
+    T = eltype(source)
+    return EnumerableDropLast{T,typeof(source)}(source, max(Int(n), 0))
+end
+
+Base.IteratorSize(::Type{EnumerableDropLast{T,S}}) where {T,S} = haslength(S)
+
+Base.eltype(::Type{EnumerableDropLast{T,S}}) where {T,S} = T
+
+Base.length(iter::EnumerableDropLast) = max(length(iter.source) - iter.n, 0)
+
+# Stays `n` elements behind the source: an element is only emitted once `n`
+# further elements have been read, which proves it is not one of the last `n`.
+function Base.iterate(iter::EnumerableDropLast{T,S}) where {T,S}
+    buffer = T[]
+    source_state = _NotStarted()
+
+    while length(buffer) < iter.n
+        ret = _iterate_from(iter.source, source_state)
+        # Fewer than n elements in total, so every one of them is dropped.
+        ret === nothing && return nothing
+        push!(buffer, ret[1])
+        source_state = ret[2]
+    end
+
+    return _drop_last_next(iter, buffer, source_state)
+end
+
+function Base.iterate(iter::EnumerableDropLast, state)
+    return _drop_last_next(iter, state.buffer, state.state)
+end
+
+function _drop_last_next(iter::EnumerableDropLast, buffer, source_state)
+    ret = _iterate_from(iter.source, source_state)
+    ret === nothing && return nothing
+
+    push!(buffer, ret[1])
+    element = popfirst!(buffer)
+
+    return element, (buffer=buffer, state=ret[2])
+end