File Coverage

lib/CSV/LINQ.pm

Criterion	Covered	Total	%
statement	456	733	62.2
branch	137	330	41.5
condition	24	72	33.3
subroutine	74	108	68.5
pod	54	55	98.1
total	745	1298	57.4

package CSV::LINQ;
###############################################################################
# CSV::LINQ - LINQ-style query interface for CSV files
# Compatible: Perl 5.005_03 and later
# Platform  : Windows / UNIX
###############################################################################

use strict;
BEGIN { if ($] < 5.006 && !defined(&warnings::import)) { $INC{'warnings.pm'} = 'stub'; eval 'package warnings; sub import {}' } }
use warnings;
local $^W = 1;

BEGIN { pop @INC if $INC[-1] eq '.' }

use vars qw($VERSION $_fh_seq);
$VERSION = '1.00';
$VERSION = $VERSION;
$_fh_seq = 0;

###############################################################################
# Internal file-handle helper
###############################################################################

# _open_fh - open a file for reading ('<') or writing ('>') and return
# the glob name string.  Works on Perl 5.005_03 and all later versions.
#
# Always uses a unique numbered package glob (CSV::LINQ::FH::H) so
# that concurrent iterators each get their own IO slot.
#
# $raw: if true, binmode is called (raw bytes).
#       Pass 0 for CSV, where OS-level \r\n->\n conversion is desired.
sub _open_fh {
    my($mode, $file, $raw) = @_;
    $_fh_seq++;
    my $seq = $_fh_seq;
    my $fhn = "CSV::LINQ::FH::H${seq}";
    my $arg = ($mode eq '>') ? ">$file" : "< $file";
    { no strict 'refs'; open($fhn, $arg) or die "Cannot open '$file': $!\n" }
    if ($raw) { no strict 'refs'; binmode(*{$fhn}) }
    return $fhn;
}

###############################################################################
# Constructor
###############################################################################

sub new {
    my($class, $iter) = @_;
    return bless { _iter => $iter }, $class;
}

###############################################################################
# CSV parsing (RFC 4180 compliant)
###############################################################################

sub _parse_csv_line {
    my($line, $sep) = @_;
    $sep = ',' unless defined $sep;
    my @fields = ();
    $line =~ s{\r\n\z|\r\z|\n\z}{};
    my $pos = 0;
    my $len = length($line);
    while ($pos <= $len) {
        if ($pos < $len && substr($line, $pos, 1) eq '"') {
            $pos++;
            my $field = '';
            while ($pos < $len) {
                my $c = substr($line, $pos, 1);
                if ($c eq '"') {
                    $pos++;
                    if ($pos < $len && substr($line, $pos, 1) eq '"') {
                        $field .= '"';
                        $pos++;
                    }
                    else {
                        last;
                    }
                }
                else {
                    $field .= $c;
                    $pos++;
                }
            }
            push @fields, $field;
            $pos++ if $pos < $len && substr($line, $pos, 1) eq $sep;
        }
        else {
            my $start = $pos;
            while ($pos < $len && substr($line, $pos, 1) ne $sep) {
                $pos++;
            }
            push @fields, substr($line, $start, $pos - $start);
            $pos++;
        }
    }
    return @fields;
}

sub _format_csv_field {
    my($value, $sep) = @_;
    $sep = ',' unless defined $sep;
    $value = '' unless defined $value;
    if ($value =~ /["\n\r]/ || index($value, $sep) >= 0) {
        $value =~ s/"/""/g;
        return '"' . $value . '"';
    }
    return $value;
}

###############################################################################
# Data source methods
###############################################################################

sub From {
    my($class, $arrayref) = @_;
    die "From() requires ARRAY reference\n"
        unless ref($arrayref) eq 'ARRAY';
    my $i = 0;
    my $iter = sub {
        return undef if $i >= scalar(@{$arrayref});
        return $arrayref->[$i++];
    };
    return $class->new($iter);
}

sub FromCSV {
    my($class, $file, %opts) = @_;
    my $sep     = defined $opts{sep}     ? $opts{sep}     : ',';
    my $headers = $opts{headers};
    my $skip    = $opts{skip_header};
    my $fhn = _open_fh('<', $file, 0);
    my @cols = ();
    {
        no strict 'refs';
        if (!defined $headers) {
            my $hdr = readline(*{$fhn});
            if (defined $hdr) {
                @cols = _parse_csv_line($hdr, $sep);
            }
        }
        else {
            @cols = @{$headers};
            if ($skip) {
                readline(*{$fhn});
            }
        }
    }
    my $iter = sub {
        no strict 'refs';
        my $line = readline(*{$fhn});
        if (!defined $line) {
            close(*{$fhn});
            return undef;
        }
        $line =~ s{\r\n\z|\r\z|\n\z}{};
        return undef if $line eq '';
        my @vals = _parse_csv_line($line, $sep);
        my %rec = ();
        for my $i (0 .. $#cols) {
            $rec{ $cols[$i] } = $vals[$i];
        }
        return { %rec };
    };
    return $class->new($iter);
}

sub Range {
    my($class, $start, $count) = @_;
    my $current = $start;
    my $end     = $start + $count - 1;
    my $iter = sub {
        return undef if $current > $end;
        return $current++;
    };
    return $class->new($iter);
}

sub Empty {
    my($class) = @_;
    my $iter = sub { return undef };
    return $class->new($iter);
}

sub Repeat {
    my($class, $element, $count) = @_;
    my $i = 0;
    my $iter = sub {
        return undef if $i >= $count;
        $i++;
        return $element;
    };
    return $class->new($iter);
}

###############################################################################
# Internal iterator helpers
###############################################################################

sub _next {
    my($self) = @_;
    return $self->{_iter}->();
}

###############################################################################
# Filtering methods
###############################################################################

sub Where {
    my($self, @args) = @_;
    my $pred;
    if (ref($args[0]) eq 'CODE') {
        $pred = $args[0];
    }
    else {
        die "Where() DSL requires even number of arguments\n"
            if @args % 2 != 0;
        my %cond = @args;
        $pred = sub {
            my $rec = $_[0];
            for my $k (keys %cond) {
                return 0 unless defined $rec->{$k};
                return 0 unless $rec->{$k} eq $cond{$k};
            }
            return 1;
        };
    }
    my $source = $self;
    my $iter = sub {
        while (1) {
            my $elem = $source->_next();
            return undef unless defined $elem;
            return $elem if $pred->($elem);
        }
    };
    return ref($self)->new($iter);
}

###############################################################################
# Projection methods
###############################################################################

sub Select {
    my($self, $selector) = @_;
    my $source = $self;
    my $iter = sub {
        my $elem = $source->_next();
        return undef unless defined $elem;
        return $selector->($elem);
    };
    return ref($self)->new($iter);
}

sub SelectMany {
    my($self, $selector) = @_;
    my $source = $self;
    my @buf = ();
    my $iter = sub {
        while (1) {
            if (@buf) {
                return shift @buf;
            }
            my $elem = $source->_next();
            return undef unless defined $elem;
            my $arr = $selector->($elem);
            die "SelectMany: selector must return an ARRAY reference\n"
                unless ref($arr) eq 'ARRAY';
            push @buf, @{$arr};
        }
    };
    return ref($self)->new($iter);
}

###############################################################################
# Concatenation methods
###############################################################################

sub Concat {
    my($self, $second) = @_;
    my $source  = $self;
    my $second2 = $second;
    my $first_done = 0;
    my $iter = sub {
        unless ($first_done) {
            my $elem = $source->_next();
            if (defined $elem) {
                return $elem;
            }
            $first_done = 1;
        }
        return $second2->_next();
    };
    return ref($self)->new($iter);
}

sub Zip {
    my($self, $second, $selector) = @_;
    my $src1 = $self;
    my $src2 = $second;
    my $iter = sub {
        my $e1 = $src1->_next();
        my $e2 = $src2->_next();
        return undef unless defined $e1 && defined $e2;
        return $selector->($e1, $e2);
    };
    return ref($self)->new($iter);
}

###############################################################################
# Partitioning methods
###############################################################################

sub Take {
    my($self, $count) = @_;
    $count = 0 if $count < 0;
    my $source = $self;
    my $taken  = 0;
    my $iter = sub {
        return undef if $taken >= $count;
        my $elem = $source->_next();
        return undef unless defined $elem;
        $taken++;
        return $elem;
    };
    return ref($self)->new($iter);
}

sub Skip {
    my($self, $count) = @_;
    $count = 0 if $count < 0;
    my $source  = $self;
    my $skipped = 0;
    my $iter = sub {
        while ($skipped < $count) {
            my $elem = $source->_next();
            return undef unless defined $elem;
            $skipped++;
        }
        return $source->_next();
    };
    return ref($self)->new($iter);
}

sub TakeWhile {
    my($self, $pred) = @_;
    my $source = $self;
    my $done   = 0;
    my $iter = sub {
        return undef if $done;
        my $elem = $source->_next();
        return undef unless defined $elem;
        if ($pred->($elem)) {
            return $elem;
        }
        $done = 1;
        return undef;
    };
    return ref($self)->new($iter);
}

sub SkipWhile {
    my($self, $pred) = @_;
    my $source     = $self;
    my $skipping   = 1;
    my $iter = sub {
        while ($skipping) {
            my $elem = $source->_next();
            return undef unless defined $elem;
            unless ($pred->($elem)) {
                $skipping = 0;
                return $elem;
            }
        }
        return $source->_next();
    };
    return ref($self)->new($iter);
}

###############################################################################
# Ordering methods
###############################################################################

sub OrderBy {
    my($self, $key_sel) = @_;
    my @items = ();
    while (defined(my $e = $self->_next())) { push @items, $e }
    return CSV::LINQ::Ordered->_new_ordered([ @items ],
        [{ sel => $key_sel, dir => 1, type => 'smart' }]);
}

sub OrderByDescending {
    my($self, $key_sel) = @_;
    my @items = ();
    while (defined(my $e = $self->_next())) { push @items, $e }
    return CSV::LINQ::Ordered->_new_ordered([ @items ],
        [{ sel => $key_sel, dir => -1, type => 'smart' }]);
}

sub OrderByStr {
    my($self, $key_sel) = @_;
    my @items = ();
    while (defined(my $e = $self->_next())) { push @items, $e }
    return CSV::LINQ::Ordered->_new_ordered([ @items ],
        [{ sel => $key_sel, dir => 1, type => 'str' }]);
}

sub OrderByStrDescending {
    my($self, $key_sel) = @_;
    my @items = ();
    while (defined(my $e = $self->_next())) { push @items, $e }
    return CSV::LINQ::Ordered->_new_ordered([ @items ],
        [{ sel => $key_sel, dir => -1, type => 'str' }]);
}

sub OrderByNum {
    my($self, $key_sel) = @_;
    my @items = ();
    while (defined(my $e = $self->_next())) { push @items, $e }
    return CSV::LINQ::Ordered->_new_ordered([ @items ],
        [{ sel => $key_sel, dir => 1, type => 'num' }]);
}

sub OrderByNumDescending {
    my($self, $key_sel) = @_;
    my @items = ();
    while (defined(my $e = $self->_next())) { push @items, $e }
    return CSV::LINQ::Ordered->_new_ordered([ @items ],
        [{ sel => $key_sel, dir => -1, type => 'num' }]);
}

sub Reverse {
    my($self) = @_;
    my @items = ();
    while (defined(my $e = $self->_next())) { push @items, $e }
    my @rev = reverse @items;
    return ref($self)->new(ref($self)->From([ @rev ])->{_iter});
}

###############################################################################
# Internal sort helpers
###############################################################################

# _extract_key($raw_value, $type) - normalise one sort key
#
# Returns a scalar for 'num'/'str', or a two-element arrayref [flag, value]
# for 'smart':
#   [0, $numeric_val]  - key is numeric
#   [1, $string_val ]  - key is string
sub _extract_key {
    my($val, $type) = @_;
    $val = '' unless defined $val;
    if ($type eq 'num') {
        return defined($val) && length($val) ? $val + 0 : 0;
    }
    elsif ($type eq 'str') {
        return "$val";
    }
    else {
        # smart: detect whether value looks like a number
        my $t = $val;
        $t =~ s{^\s+}{};
        $t =~ s{\s+$}{};
        if ($t =~ /^[+-]?(?:\d+\.?\d*|\d*\.\d+)(?:[eE][+-]?\d+)?$/) {
            return [0, $t + 0];
        }
        else {
            return [1, "$val"];
        }
    }
}

# _compare_keys($ka, $kb, $type) - compare two extracted keys
sub _compare_keys {
    my($ka, $kb, $type) = @_;
    if ($type eq 'num') {
        return $ka <=> $kb;
    }
    elsif ($type eq 'str') {
        return $ka cmp $kb;
    }
    else {
        # smart: both are [flag, value] arrayrefs
        my $fa = $ka->[0]; my $va = $ka->[1];
        my $fb = $kb->[0]; my $vb = $kb->[1];
        if    ($fa == 0 && $fb == 0) { return $va <=> $vb }
        elsif ($fa == 1 && $fb == 1) { return $va cmp $vb }
        else                         { return $fa <=> $fb  }
    }
}

# _perform_sort($items_aref, $specs_aref) - stable multi-key sort
#
# Schwartzian Transform:
#   1. Decorate each element: [ orig_index, [key1..keyN], item ]
#   2. Sort by keys in sequence; original index as final tie-breaker (stability)
#   3. Undecorate
sub _perform_sort {
    my($items, $specs) = @_;

    # Step 1: decorate
    my @decorated = map {
        my $idx  = $_;
        my $item = $items->[$idx];
        my @keys = map { _extract_key($_->{sel}->($item), $_->{type}) } @{$specs};
        [$idx, [ @keys ], $item]
    } 0 .. $#{$items};

    # Step 2: sort
    my @sorted_dec = sort {
        my $r = 0;
        for my $i (0 .. $#{$specs}) {
            my $cmp = _compare_keys($a->[1][$i], $b->[1][$i], $specs->[$i]{type});
            if ($specs->[$i]{dir} < 0) { $cmp = -$cmp }
            if ($cmp != 0) { $r = $cmp; last }
        }
        $r != 0 ? $r : ($a->[0] <=> $b->[0]);
    } @decorated;

    # Step 3: undecorate
    return map { $_->[2] } @sorted_dec;
}

###############################################################################
# Ordered sub-class (ThenBy support)
###############################################################################

package CSV::LINQ::Ordered;
use vars qw(@ISA);
@ISA = ('CSV::LINQ');

# _new_ordered($items_aref, $specs_aref) - internal constructor
#
# $specs_aref is an arrayref of sort-spec hashrefs:
#   { sel  => $code_ref,           # key selector: ($item) -> $key
#     dir  => 1 or -1,             # 1 = ascending, -1 = descending
#     type => 'smart'|'str'|'num'  # comparison family
#   }
#
# Uses _factory so each _next() call draws from a fresh sorted iterator,
# enabling re-iteration of Ordered objects.
sub _new_ordered {
    my($class, $items, $specs) = @_;
    return bless {
        _items   => $items,
        _specs   => $specs,
        _factory => sub {
            my @sorted = CSV::LINQ::_perform_sort($items, $specs);
            my $i = 0;
            return sub { $i < scalar(@sorted) ? $sorted[$i++] : undef };
        },
    }, $class;
}

# Override _next() to use _factory if available
# When the iterator signals end-of-sequence (undef), clear {_iter} so
# the next call to _next() rebuilds a fresh iterator from {_factory},
# enabling re-iteration of the same Ordered object.
sub _next {
    my($self) = @_;
    unless (exists $self->{_iter}) {
        $self->{_iter} = $self->{_factory}->();
    }
    my $val = $self->{_iter}->();
    unless (defined $val) {
        delete $self->{_iter};
    }
    return $val;
}

# _thenby - shared implementation for all ThenBy* variants
#
# Non-destructive: builds a new spec list and returns a new
# CSV::LINQ::Ordered object. The original object is unchanged.
sub _thenby {
    my($self, $key_sel, $dir, $type) = @_;
    my @new_specs = (@{$self->{_specs}}, { sel => $key_sel, dir => $dir, type => $type });
    return CSV::LINQ::Ordered->_new_ordered($self->{_items}, [ @new_specs ]);
}

sub ThenBy              { my($s, $k)=@_; $s->_thenby($k,  1, 'smart') }
sub ThenByDescending    { my($s, $k)=@_; $s->_thenby($k, -1, 'smart') }
sub ThenByStr           { my($s, $k)=@_; $s->_thenby($k,  1, 'str')   }
sub ThenByStrDescending { my($s, $k)=@_; $s->_thenby($k, -1, 'str')   }
sub ThenByNum           { my($s, $k)=@_; $s->_thenby($k,  1, 'num')   }
sub ThenByNumDescending { my($s, $k)=@_; $s->_thenby($k, -1, 'num')   }

package CSV::LINQ;


###############################################################################
# Grouping methods
###############################################################################

sub GroupBy {
    my($self, $key_sel, $elem_sel) = @_;
    my %groups  = ();
    my @keys    = ();
    while (defined(my $e = $self->_next())) {
        my $k = $key_sel->($e);
        $k = '' unless defined $k;
        unless (exists $groups{$k}) {
            push @keys, $k;
            $groups{$k} = [];
        }
        if (defined $elem_sel) {
            push @{ $groups{$k} }, $elem_sel->($e);
        }
        else {
            push @{ $groups{$k} }, $e;
        }
    }
    my @result = ();
    for my $k (@keys) {
        push @result, { Key => $k, Elements => $groups{$k} };
    }
    return ref($self)->From([ @result ]);
}

###############################################################################
# Set operations
###############################################################################

sub Distinct {
    my($self, $comparer) = @_;
    my $source = $self;
    my %seen   = ();
    my $iter = sub {
        while (1) {
            my $elem = $source->_next();
            return undef unless defined $elem;
            my $key = defined $comparer ? $comparer->($elem) : $elem;
            $key = '' unless defined $key;
            next if $seen{$key};
            $seen{$key} = 1;
            return $elem;
        }
    };
    return ref($self)->new($iter);
}

sub Union {
    my($self, $second, $comparer) = @_;
    return $self->Concat($second)->Distinct($comparer);
}

sub Intersect {
    my($self, $second, $comparer) = @_;
    my %in2 = ();
    while (defined(my $e = $second->_next())) {
        my $k = defined $comparer ? $comparer->($e) : $e;
        $k = '' unless defined $k;
        $in2{$k} = 1;
    }
    my $source = $self;
    my %seen   = ();
    my $iter = sub {
        while (1) {
            my $elem = $source->_next();
            return undef unless defined $elem;
            my $k = defined $comparer ? $comparer->($elem) : $elem;
            $k = '' unless defined $k;
            next unless $in2{$k};
            next if $seen{$k};
            $seen{$k} = 1;
            return $elem;
        }
    };
    return ref($self)->new($iter);
}

sub Except {
    my($self, $second, $comparer) = @_;
    my %in2 = ();
    while (defined(my $e = $second->_next())) {
        my $k = defined $comparer ? $comparer->($e) : $e;
        $k = '' unless defined $k;
        $in2{$k} = 1;
    }
    my $source = $self;
    my %seen   = ();
    my $iter = sub {
        while (1) {
            my $elem = $source->_next();
            return undef unless defined $elem;
            my $k = defined $comparer ? $comparer->($elem) : $elem;
            $k = '' unless defined $k;
            next if $in2{$k};
            next if $seen{$k};
            $seen{$k} = 1;
            return $elem;
        }
    };
    return ref($self)->new($iter);
}

###############################################################################
# Join operations
###############################################################################

sub Join {
    my($self, $inner, $outer_key, $inner_key, $result_sel) = @_;
    my %lookup = ();
    while (defined(my $e = $inner->_next())) {
        my $k = $inner_key->($e);
        $k = '' unless defined $k;
        $lookup{$k} = [] unless exists $lookup{$k};
        push @{ $lookup{$k} }, $e;
    }
    my $source = $self;
    my @buf    = ();
    my $iter = sub {
        while (1) {
            if (@buf) {
                return shift @buf;
            }
            my $outer = $source->_next();
            return undef unless defined $outer;
            my $k = $outer_key->($outer);
            $k = '' unless defined $k;
            next unless exists $lookup{$k};
            for my $inner_elem (@{ $lookup{$k} }) {
                push @buf, $result_sel->($outer, $inner_elem);
            }
        }
    };
    return ref($self)->new($iter);
}

sub GroupJoin {
    my($self, $inner, $outer_key, $inner_key, $result_sel) = @_;
    my %lookup = ();
    while (defined(my $e = $inner->_next())) {
        my $k = $inner_key->($e);
        $k = '' unless defined $k;
        $lookup{$k} = [] unless exists $lookup{$k};
        push @{ $lookup{$k} }, $e;
    }
    my $source = $self;
    my $iter = sub {
        my $outer = $source->_next();
        return undef unless defined $outer;
        my $k = $outer_key->($outer);
        $k = '' unless defined $k;
        my $group = exists $lookup{$k} ? $lookup{$k} : [];
        my $inner_query = ref($self)->From([ @{$group} ]);
        return $result_sel->($outer, $inner_query);
    };
    return ref($self)->new($iter);
}

###############################################################################
# Quantifier methods
###############################################################################

sub All {
    my($self, $pred) = @_;
    while (defined(my $e = $self->_next())) {
        return 0 unless $pred->($e);
    }
    return 1;
}

sub Any {
    my($self, $pred) = @_;
    if (defined $pred) {
        while (defined(my $e = $self->_next())) {
            return 1 if $pred->($e);
        }
        return 0;
    }
    else {
        return defined($self->_next()) ? 1 : 0;
    }
}

sub Contains {
    my($self, $value, $comparer) = @_;
    while (defined(my $e = $self->_next())) {
        if (defined $comparer) {
            return 1 if $comparer->($e, $value);
        }
        else {
            if (!defined $value) {
                return 1 unless defined $e;
            }
            elsif (defined $e && $e eq $value) {
                return 1;
            }
        }
    }
    return 0;
}

sub SequenceEqual {
    my($self, $second, $comparer) = @_;
    while (1) {
        my $e1 = $self->_next();
        my $e2 = $second->_next();
        if (!defined $e1 && !defined $e2) {
            return 1;
        }
        return 0 if !defined $e1 || !defined $e2;
        if (defined $comparer) {
            return 0 unless $comparer->($e1, $e2);
        }
        else {
            return 0 unless $e1 eq $e2;
        }
    }
}

###############################################################################
# Element access methods
###############################################################################

sub First {
    my($self, $pred) = @_;
    while (defined(my $e = $self->_next())) {
        if (!defined $pred || $pred->($e)) {
            return $e;
        }
    }
    if (defined $pred) {
        die "No element satisfies the condition\n";
    }
    die "Sequence contains no elements\n";
}

sub FirstOrDefault {
    my($self, $pred, $default) = @_;
    if (ref($pred) eq 'CODE') {
        while (defined(my $e = $self->_next())) {
            return $e if $pred->($e);
        }
        return $default;
    }
    else {
        $default = $pred;
        my $e = $self->_next();
        return defined $e ? $e : $default;
    }
}

sub Last {
    my($self, $pred) = @_;
    my $found;
    my $has = 0;
    while (defined(my $e = $self->_next())) {
        if (!defined $pred || $pred->($e)) {
            $found = $e;
            $has   = 1;
        }
    }
    if ($has) {
        return $found;
    }
    if (defined $pred) {
        die "No element satisfies the condition\n";
    }
    die "Sequence contains no elements\n";
}

sub LastOrDefault {
    my($self, $pred) = @_;
    my $found;
    my $has = 0;
    while (defined(my $e = $self->_next())) {
        if (!defined $pred || $pred->($e)) {
            $found = $e;
            $has   = 1;
        }
    }
    return $has ? $found : undef;
}

sub Single {
    my($self, $pred) = @_;
    my $found;
    my $count = 0;
    while (defined(my $e = $self->_next())) {
        if (!defined $pred || $pred->($e)) {
            $found = $e;
            $count++;
            die "Sequence contains more than one element\n"
                if $count > 1;
        }
    }
    die "Sequence contains no elements\n" if $count == 0;
    return $found;
}

sub SingleOrDefault {
    my($self, $pred) = @_;
    my $found;
    my $count = 0;
    while (defined(my $e = $self->_next())) {
        if (!defined $pred || $pred->($e)) {
            $found = $e;
            $count++;
            return undef if $count > 1;
        }
    }
    return $count == 1 ? $found : undef;
}

sub ElementAt {
    my($self, $index) = @_;
    die "ElementAt: index out of range\n" if $index < 0;
    my $i = 0;
    while (defined(my $e = $self->_next())) {
        return $e if $i == $index;
        $i++;
    }
    die "ElementAt: index out of range\n";
}

sub ElementAtOrDefault {
    my($self, $index) = @_;
    return undef if $index < 0;
    my $i = 0;
    while (defined(my $e = $self->_next())) {
        return $e if $i == $index;
        $i++;
    }
    return undef;
}

###############################################################################
# Aggregation methods
###############################################################################

sub Count {
    my($self, $pred) = @_;
    my $n = 0;
    while (defined(my $e = $self->_next())) {
        if (!defined $pred || $pred->($e)) {
            $n++;
        }
    }
    return $n;
}

sub Sum {
    my($self, $selector) = @_;
    my $total = 0;
    while (defined(my $e = $self->_next())) {
        my $v = defined $selector ? $selector->($e) : $e;
        $total += (defined $v ? $v : 0);
    }
    return $total;
}

sub Min {
    my($self, $selector) = @_;
    my $min;
    while (defined(my $e = $self->_next())) {
        my $v = defined $selector ? $selector->($e) : $e;
        next unless defined $v;
        $min = $v if !defined $min || $v < $min;
    }
    return $min;
}

sub Max {
    my($self, $selector) = @_;
    my $max;
    while (defined(my $e = $self->_next())) {
        my $v = defined $selector ? $selector->($e) : $e;
        next unless defined $v;
        $max = $v if !defined $max || $v > $max;
    }
    return $max;
}

sub Average {
    my($self, $selector) = @_;
    my $sum = 0;
    my $n   = 0;
    while (defined(my $e = $self->_next())) {
        my $v = defined $selector ? $selector->($e) : $e;
        $sum += (defined $v ? $v : 0);
        $n++;
    }
    die "Sequence contains no elements\n" if $n == 0;
    return $sum / $n;
}

sub AverageOrDefault {
    my($self, $selector) = @_;
    my $sum = 0;
    my $n   = 0;
    while (defined(my $e = $self->_next())) {
        my $v = defined $selector ? $selector->($e) : $e;
        $sum += (defined $v ? $v : 0);
        $n++;
    }
    return $n == 0 ? undef : $sum / $n;
}

sub Aggregate {
    my($self, @args) = @_;
    my($seed, $func, $result_sel);
    if (@args == 1) {
        $func = $args[0];
        my $first = $self->_next();
        return undef unless defined $first;
        $seed = $first;
    }
    elsif (@args == 2) {
        ($seed, $func) = @args;
    }
    else {
        ($seed, $func, $result_sel) = @args;
    }
    my $acc = $seed;
    while (defined(my $e = $self->_next())) {
        $acc = $func->($acc, $e);
    }
    return defined $result_sel ? $result_sel->($acc) : $acc;
}

###############################################################################
# Conversion methods
###############################################################################

sub ToArray {
    my($self) = @_;
    my @result = ();
    while (defined(my $e = $self->_next())) {
        push @result, $e;
    }
    return @result;
}

sub ToList {
    my($self) = @_;
    my @result = ();
    while (defined(my $e = $self->_next())) {
        push @result, $e;
    }
    return [ @result ];
}

sub DefaultIfEmpty {
    my($self, $default) = @_;
    my $source  = $self;
    my $started = 0;
    my $done    = 0;
    my $iter = sub {
        if (!$started) {
            my $elem = $source->_next();
            $started = 1;
            unless (defined $elem) {
                unless ($done) {
                    $done = 1;
                    return $default;
                }
                return undef;
            }
            return $elem;
        }
        return undef if $done;
        return $source->_next();
    };
    return ref($self)->new($iter);
}

sub ToDictionary {
    my($self, $key_sel, $val_sel) = @_;
    my %dict = ();
    while (defined(my $e = $self->_next())) {
        my $k = $key_sel->($e);
        $k = '' unless defined $k;
        my $v = defined $val_sel ? $val_sel->($e) : $e;
        $dict{$k} = $v;
    }
    return { %dict };
}

sub ToLookup {
    my($self, $key_sel, $val_sel) = @_;
    my %lookup = ();
    my @keys   = ();
    while (defined(my $e = $self->_next())) {
        my $k = $key_sel->($e);
        $k = '' unless defined $k;
        unless (exists $lookup{$k}) {
            push @keys, $k;
            $lookup{$k} = [];
        }
        my $v = defined $val_sel ? $val_sel->($e) : $e;
        push @{ $lookup{$k} }, $v;
    }
    return { %lookup };
}

sub ToCSV {
    my($self, $file, %opts) = @_;
    my $sep       = defined $opts{sep}       ? $opts{sep}       : ',';
    my $headers   = defined $opts{headers}   ? $opts{headers}
                  : defined $opts{label_order} ? $opts{label_order}
                  : undef;
    my $no_header = $opts{no_header};
    my @items = ();
    while (defined(my $e = $self->_next())) {
        push @items, $e;
    }
    my $fhn = _open_fh('>', $file, 0);
    unless ($no_header) {
        my @cols = ();
        if (defined $headers) {
            @cols = @{$headers};
        }
        elsif (@items && ref($items[0]) eq 'HASH') {
            @cols = sort keys %{ $items[0] };
        }
        if (@cols) {
            no strict 'refs';
            print {*{$fhn}}
                join($sep, map { _format_csv_field($_, $sep) } @cols) . "\n";
        }
        if (!@cols && @items && ref($items[0]) ne 'HASH') {
            # scalar sequence - no header
        }
        else {
            for my $item (@items) {
                no strict 'refs';
                if (ref($item) eq 'HASH') {
                    print {*{$fhn}}
                        join($sep, map {
                            _format_csv_field($item->{$_}, $sep)
                        } @cols) . "\n";
                }
                else {
                    print {*{$fhn}}
                        _format_csv_field($item, $sep) . "\n";
                }
            }
            no strict 'refs';
            close(*{$fhn});
            return 1;
        }
    }
    for my $item (@items) {
        no strict 'refs';
        if (ref($item) eq 'HASH') {
            my @cols = sort keys %{$item};
            print {*{$fhn}}
                join($sep, map {
                    _format_csv_field($item->{$_}, $sep)
                } @cols) . "\n";
        }
        else {
            print {*{$fhn}}
                _format_csv_field($item, $sep) . "\n";
        }
    }
    {
        no strict 'refs';
        close(*{$fhn});
    }
    return 1;
}

###############################################################################
# Utility methods
###############################################################################

sub ForEach {
    my($self, $action) = @_;
    while (defined(my $e = $self->_next())) {
        $action->($e);
    }
    return;
}

1;

__END__