DBIx::Class::ResultSet - Represents a query used for fetching a set of results.

  my $users_rs   = $schema->resultset('User');
  while( $user = $users_rs->next) {
    print $user->username;
  }

  my $registered_users_rs   = $schema->resultset('User')->search({ registered => 1 });
  my @cds_in_2005 = $schema->resultset('CD')->search({ year => 2005 })->all();

  my $users_rs = $schema->resultset('User');

  sub get_data {
    my $self = shift;
    my $request = $self->get_request; # Get a request object somehow.
    my $schema = $self->get_schema;   # Get the DBIC schema object somehow.

    my $cd_rs = $schema->resultset('CD')->search({
      title => $request->param('title'),
      year => $request->param('year'),
    });

    $cd_rs = $self->apply_security_policy( $cd_rs );

    return $cd_rs->all();
  }

  sub apply_security_policy {
    my $self = shift;
    my ($rs) = @_;

    return $rs->search({
      subversive => 0,
    });
  }

  # Don't hit the DB yet.
  my $cd_rs = $schema->resultset('CD')->search({
    title => 'something',
    year => 2009,
  });

  # Each of these hits the DB individually.
  my $count = $cd_rs->count;
  my $most_recent = $cd_rs->get_column('date_released')->max();
  my @records = $cd_rs->all;

  $cd_rs->delete();

  $cd_rs->create({ artist => 'Fred' });

  $schema->resultset('CD')->create({
    title => 'something',
    year => 2009,
    artist => 'Fred'
  });

  my $rs = $schema->resultset('CD')->search({ title => '100th Window' });

  my $cd = $schema->resultset('CD')->new({ title => 'Spoon' });

  my @cds    = $cd_rs->search({ year => 2001 }); # "... WHERE year = 2001"
  my $new_rs = $cd_rs->search({ year => 2005 });

  my $new_rs = $cd_rs->search([ { year => 2005 }, { year => 2004 } ]);
                 # year = 2005 OR year = 2004

  # "SELECT name, artistid FROM $artist_table"
  my @all_artists = $schema->resultset('Artist')->search(undef, {
    columns => [qw/name artistid/],
  });

  my @cds   = $cd_rs->search_literal('year = ? AND title = ?', qw/2001 Reload/);
  my $newrs = $artist_rs->search_literal('name = ?', 'Metallica');

  my @cds = $cd_rs->search_literal('cdid = ? AND (artist = ? OR artist = ?)', (2, 1, 2));
  my @cds = $cd_rs->search(\[ 'cdid = ? AND (artist = ? OR artist = ?)', [ 'cdid', 2 ], [ 'artist', 1 ], [ 'artist', 2 ] ]);

  my $cd = $schema->resultset('CD')->find(5);

  my $cd = $schema->resultset('CD')->find(
    {
      artist => 'Massive Attack',
      title  => 'Mezzanine',
    },
    { key => 'cd_artist_title' }
  );

  $new_rs = $cd_rs->search_related('artist', {
    name => 'Emo-R-Us',
  });

  my $cd = $schema->resultset('CD')->single({ year => 2001 });

  my $max_length = $rs->get_column('length')->max;

  # WHERE title LIKE '%blue%'
  $cd_rs = $rs->search_like({ title => '%blue%'});

  ->search_like({ foo => 'bar' });

  # Becomes

  ->search({ foo => { like => 'bar' } });

  my ($one, $two, $three) = $rs->slice(0, 2);

  my $rs = $schema->resultset('CD')->search;
  while (my $cd = $rs->next) {
    print $cd->title;
  }

  ->search( { amount => $some_rs->count_rs->as_query } )

  my $Artist_rs = $schema->resultset("Artist");

  ## Void Context Example
  $Artist_rs->populate([
     { artistid => 4, name => 'Manufactured Crap', cds => [
        { title => 'My First CD', year => 2006 },
        { title => 'Yet More Tweeny-Pop crap', year => 2007 },
      ],
     },
     { artistid => 5, name => 'Angsty-Whiny Girl', cds => [
        { title => 'My parents sold me to a record company', year => 2005 },
        { title => 'Why Am I So Ugly?', year => 2006 },
        { title => 'I Got Surgery and am now Popular', year => 2007 }
      ],
     },
  ]);

  ## Array Context Example
  my ($ArtistOne, $ArtistTwo, $ArtistThree) = $Artist_rs->populate([
    { name => "Artist One"},
    { name => "Artist Two"},
    { name => "Artist Three", cds=> [
    { title => "First CD", year => 2007},
    { title => "Second CD", year => 2008},
  ]}
  ]);

  print $ArtistOne->name; ## response is 'Artist One'
  print $ArtistThree->cds->count ## reponse is '2'

  $Arstist_rs->populate([
    [qw/artistid name/],
    [100, 'A Formally Unknown Singer'],
    [101, 'A singer that jumped the shark two albums ago'],
    [102, 'An actually cool singer'],
  ]);

  my $artist = $schema->resultset('Artist')->find_or_new(
    { artist => 'fred' }, { key => 'artists' });

  $cd->cd_to_producer->find_or_new({ producer => $producer },
                                   { key => 'primary });

  $person_rs->create({
    name=>"Some Person",
    email=>"somebody@someplace.com"
  });

  $artist_rs->create(
     { artistid => 4, name => 'Manufactured Crap', cds => [
        { title => 'My First CD', year => 2006 },
        { title => 'Yet More Tweeny-Pop crap', year => 2007 },
      ],
     },
  );

  $cd_rs->create({
    title=>"Music for Silly Walks",
    year=>2000,
    artist => {
      name=>"Silly Musician",
    }
  });

  $cd->cd_to_producer->find_or_create({ producer => $producer },
                                      { key => 'primary' });

  my $cd = $schema->resultset('CD')->find_or_create({
    cdid   => 5,
    artist => 'Massive Attack',
    title  => 'Mezzanine',
    year   => 2005,
  });

  my $cd = $schema->resultset('CD')->find_or_create(
    {
      artist => 'Massive Attack',
      title  => 'Mezzanine',
    },
    { key => 'cd_artist_title' }
  );

  $resultset->update_or_create({ col => $val, ... });

  # In your application
  my $cd = $schema->resultset('CD')->update_or_create(
    {
      artist => 'Massive Attack',
      title  => 'Mezzanine',
      year   => 1998,
    },
    { key => 'cd_artist_title' }
  );

  $cd->cd_to_producer->update_or_create({
    producer => $producer,
    name => 'harry',
  }, {
    key => 'primary',
  });

  $resultset->update_or_new({ col => $val, ... });

  # In your application
  my $cd = $schema->resultset('CD')->update_or_new(
    {
      artist => 'Massive Attack',
      title  => 'Mezzanine',
      year   => 1998,
    },
    { key => 'cd_artist_title' }
  );

  if ($cd->in_storage) {
      # the cd was updated
  }
  else {
      # the cd is not yet in the database, let's insert it
      $cd->insert;
  }

  $artist_rs = $schema->resultset('CD')->related_resultset('Artist');

  # in a result set class
  sub modified_by {
    my ($self, $user) = @_;

    my $me = $self->current_source_alias;

    return $self->search(
      "$me.modified" => $user->id,
    );
  }

 my $rs = $schema->resultset('Bar')->search({'x.name' => 'abc'},{ join => 'x' });

 # 'x' now pollutes the query namespace

 # So the following works as expected
 my $ok_rs = $rs->search({'x.other' => 1});

 # But this doesn't: instead of finding a 'Bar' related to two x rows (abc and
 # def) we look for one row with contradictory terms and join in another table
 # (aliased 'x_2') which we never use
 my $broken_rs = $rs->search({'x.name' => 'def'});

 my $rs2 = $rs->as_subselect_rs;

 # doesn't work - 'x' is no longer accessible in $rs2, having been sealed away
 my $not_joined_rs = $rs2->search({'x.other' => 1});

 # works as expected: finds a 'table' row related to two x rows (abc and def)
 my $correctly_joined_rs = $rs2->search({'x.name' => 'def'});

 my $rs = $schema->resultset('Bar')->search(undef, {
   group_by => [qw{ id foo_id baz_id }],
 })->as_subselect_rs->search(undef, {
   columns => [qw{ id foo_id }]
 });

 For descending order:

  order_by => { -desc => [qw/col1 col2 col3/] }

 For explicit ascending order:

  order_by => { -asc => 'col' }

    columns => [ 'foo', { bar => 'baz' } ]

    select => [qw/foo baz/],
    as => [qw/foo bar/]

  $schema->resultset('CD')->search(undef, {
    '+columns' => ['artist.name'],
    join => ['artist']
  });

  $rs = $schema->resultset('Employee')->search(undef, {
    select => [
      'name',
      { count => 'employeeid' },
      { max => { length => 'name' }, -as => 'longest_name' }
    ]
  });

  # Equivalent SQL
  SELECT name, COUNT( employeeid ), MAX( LENGTH( name ) ) AS longest_name FROM employee

  $rs = $schema->resultset('Employee')->search(undef, {
    select => [
      'name',
      { count => 'employeeid' },
      { max => { length => 'name' }, -as => 'longest_name' }
    ],
    as => [qw/
      name
      employee_count
      max_name_length
    /],
  });

  my $name = $employee->name();

  my $employee_count = $employee->get_column('employee_count');

  # Get CDs by Nine Inch Nails
  my $rs = $schema->resultset('CD')->search(
    { 'artist.name' => 'Nine Inch Nails' },
    { join => 'artist' }
  );

  package MyApp::Schema::Track;
  use base qw/DBIx::Class/;
  __PACKAGE__->table('track');
  __PACKAGE__->add_columns(qw/trackid cd position title/);
  __PACKAGE__->set_primary_key('trackid');
  __PACKAGE__->belongs_to(cd => 'MyApp::Schema::CD');
  1;

  # In your application
  my $rs = $schema->resultset('Artist')->search(
    { 'track.title' => 'Teardrop' },
    {
      join     => { cd => 'track' },
      order_by => 'artist.name',
    }
  );

  # Get CDs from 1984 with a 'Foo' track
  my $rs = $schema->resultset('CD')->search(
    {
      'me.year' => 1984,
      'tracks.name' => 'Foo'
    },
    { join => 'tracks' }
  );

  my $rs = $schema->resultset('Artist')->search({
    'cds.title'   => 'Down to Earth',
    'cds_2.title' => 'Popular',
  }, {
    join => [ qw/cds cds/ ],
  });

  my $rs = $schema->resultset('Tag')->search(
    undef,
    {
      prefetch => {
        cd => 'artist'
      }
    }
  );

  SELECT tag.*, cd.*, artist.* FROM tag
  JOIN cd ON tag.cd = cd.cdid
  JOIN artist ON cd.artist = artist.artistid

 # Assuming:
 My::Schema::CD->belongs_to( artist      => 'My::Schema::Artist'     );
 My::Schema::CD->might_have( liner_note  => 'My::Schema::LinerNotes' );
 My::Schema::CD->has_one(    cover_image => 'My::Schema::Artwork'    );
 My::Schema::CD->has_many(   tracks      => 'My::Schema::Track'      );

 My::Schema::Artist->belongs_to( record_label => 'My::Schema::RecordLabel' );

 My::Schema::Track->has_many( guests => 'My::Schema::Guest' );




 my $rs = $schema->resultset('CD')->search(
   undef,
   {
     prefetch => [
       { artist => 'record_label'},  # belongs_to => belongs_to
       'liner_note',                 # might_have
       'cover_image',                # has_one
       { tracks => 'guests' },       # has_many => has_many
     ]
   }
 );

 SELECT cd.*, artist.*, record_label.*, liner_note.*, cover_image.*,
        tracks.*, guests.*
   FROM cd me
   JOIN artist artist
     ON artist.artistid = me.artistid
   JOIN record_label record_label
     ON record_label.labelid = artist.labelid
   LEFT JOIN track tracks
     ON tracks.cdid = me.cdid
   LEFT JOIN guest guests
     ON guests.trackid = track.trackid
   LEFT JOIN liner_notes liner_note
     ON liner_note.cdid = me.cdid
   JOIN cd_artwork cover_image
     ON cover_image.cdid = me.cdid
 ORDER BY tracks.cd

 my $rs = $schema->resultset('CD')->search(
   undef,
   {
     prefetch => [
       'tracks',                         # has_many
       { cd_to_producer => 'producer' }, # has_many => belongs_to (i.e. m2m)
     ]
   }
 );

 Prefetching multiple has_many rels tracks and cd_to_producer at top
 level will explode the number of row objects retrievable via ->next
 or ->all. Use at your own risk.

  my $rs = $schema->resultset('CD')->search(
   {'record_label.name' => 'Music Product Ltd.'},
   {
     join     => {artist => 'record_label'},
     prefetch => 'artist',
   }
 );

  my $rs = $schema->resultset('CD')->search(
   undef,
   {
     select   => ['cd.title'],
     as       => ['cd_title'],
     prefetch => 'artist',
   }
 );

  group_by => [qw/ column1 column2 ... /]

  having => { 'count_employee' => { '>=', 100 } }

  having => \[ 'count(employee) >= ?', [ count => 100 ] ]

  # only return rows WHERE deleted IS NULL for all searches
  __PACKAGE__->resultset_attributes({ where => { deleted => undef } }); )

  my $resultset = $schema->resultset('Artist')->search( undef, { cache => 1 } );

  while( my $artist = $resultset->next ) {
    ... do stuff ...
  }

  $rs->first; # without cache, this would issue a query

package DBIx::Class::ResultSet;

use strict;
use warnings;
use base qw/DBIx::Class/;
use DBIx::Class::Carp;
use DBIx::Class::Exception;
use DBIx::Class::ResultSetColumn;
use Scalar::Util qw/blessed weaken/;
use Try::Tiny;

# not importing first() as it will clash with our own method
use List::Util ();

BEGIN {
  # De-duplication in _merge_attr() is disabled, but left in for reference
  # (the merger is used for other things that ought not to be de-duped)
  *__HM_DEDUP = sub () { 0 };
}

use namespace::clean;

use overload
        '0+'     => "count",
        'bool'   => "_bool",
        fallback => 1;

__PACKAGE__->mk_group_accessors('simple' => qw/_result_class result_source/);

sub new {
  my $class = shift;
  return $class->new_result(@_) if ref $class;

  my ($source, $attrs) = @_;
  $source = $source->resolve
    if $source->isa('DBIx::Class::ResultSourceHandle');
  $attrs = { %{$attrs||{}} };

  if ($attrs->{page}) {
    $attrs->{rows} ||= 10;
  }

  $attrs->{alias} ||= 'me';

  my $self = bless {
    result_source => $source,
    cond => $attrs->{where},
    pager => undef,
    attrs => $attrs,
  }, $class;

  # if there is a dark selector, this means we are already in a
  # chain and the cleanup/sanification was taken care of by
  # _search_rs already
  $self->_normalize_selection($attrs)
    unless $attrs->{_dark_selector};

  $self->result_class(
    $attrs->{result_class} || $source->result_class
  );

  $self;
}

sub search {
  my $self = shift;
  my $rs = $self->search_rs( @_ );

  if (wantarray) {
    return $rs->all;
  }
  elsif (defined wantarray) {
    return $rs;
  }
  else {
    # we can be called by a relationship helper, which in
    # turn may be called in void context due to some braindead
    # overload or whatever else the user decided to be clever
    # at this particular day. Thus limit the exception to
    # external code calls only
    $self->throw_exception ('->search is *not* a mutator, calling it in void context makes no sense')
      if (caller)[0] !~ /^\QDBIx::Class::/;

    return ();
  }
}

sub search_rs {
  my $self = shift;

  # Special-case handling for (undef, undef).
  if ( @_ == 2 && !defined $_[1] && !defined $_[0] ) {
    @_ = ();
  }

  my $call_attrs = {};
  if (@_ > 1) {
    if (ref $_[-1] eq 'HASH') {
      # copy for _normalize_selection
      $call_attrs = { %{ pop @_ } };
    }
    elsif (! defined $_[-1] ) {
      pop @_;   # search({}, undef)
    }
  }

  # see if we can keep the cache (no $rs changes)
  my $cache;
  my %safe = (alias => 1, cache => 1);
  if ( ! List::Util::first { !$safe{$_} } keys %$call_attrs and (
    ! defined $_[0]
      or
    ref $_[0] eq 'HASH' && ! keys %{$_[0]}
      or
    ref $_[0] eq 'ARRAY' && ! @{$_[0]}
  )) {
    $cache = $self->get_cache;
  }

  my $rsrc = $self->result_source;

  my $old_attrs = { %{$self->{attrs}} };
  my $old_having = delete $old_attrs->{having};
  my $old_where = delete $old_attrs->{where};

  my $new_attrs = { %$old_attrs };

  # take care of call attrs (only if anything is changing)
  if (keys %$call_attrs) {

    my @selector_attrs = qw/select as columns cols +select +as +columns include_columns/;

    # reset the current selector list if new selectors are supplied
    if (List::Util::first { exists $call_attrs->{$_} } qw/columns cols select as/) {
      delete @{$old_attrs}{(@selector_attrs, '_dark_selector')};
    }

    # Normalize the new selector list (operates on the passed-in attr structure)
    # Need to do it on every chain instead of only once on _resolved_attrs, in
    # order to allow detection of empty vs partial 'as'
    $call_attrs->{_dark_selector} = $old_attrs->{_dark_selector}
      if $old_attrs->{_dark_selector};
    $self->_normalize_selection ($call_attrs);

    # start with blind overwriting merge, exclude selector attrs
    $new_attrs = { %{$old_attrs}, %{$call_attrs} };
    delete @{$new_attrs}{@selector_attrs};

    for (@selector_attrs) {
      $new_attrs->{$_} = $self->_merge_attr($old_attrs->{$_}, $call_attrs->{$_})
        if ( exists $old_attrs->{$_} or exists $call_attrs->{$_} );
    }

    # older deprecated name, use only if {columns} is not there
    if (my $c = delete $new_attrs->{cols}) {
      if ($new_attrs->{columns}) {
        carp "Resultset specifies both the 'columns' and the legacy 'cols' attributes - ignoring 'cols'";
      }
      else {
        $new_attrs->{columns} = $c;
      }
    }


    # join/prefetch use their own crazy merging heuristics
    foreach my $key (qw/join prefetch/) {
      $new_attrs->{$key} = $self->_merge_joinpref_attr($old_attrs->{$key}, $call_attrs->{$key})
        if exists $call_attrs->{$key};
    }

    # stack binds together
    $new_attrs->{bind} = [ @{ $old_attrs->{bind} || [] }, @{ $call_attrs->{bind} || [] } ];
  }


  # rip apart the rest of @_, parse a condition
  my $call_cond = do {

    if (ref $_[0] eq 'HASH') {
      (keys %{$_[0]}) ? $_[0] : undef
    }
    elsif (@_ == 1) {
      $_[0]
    }
    elsif (@_ % 2) {
      $self->throw_exception('Odd number of arguments to search')
    }
    else {
      +{ @_ }
    }

  } if @_;

  if( @_ > 1 and ! $rsrc->result_class->isa('DBIx::Class::CDBICompat') ) {
    carp_unique 'search( %condition ) is deprecated, use search( \%condition ) instead';
  }

  for ($old_where, $call_cond) {
    if (defined $_) {
      $new_attrs->{where} = $self->_stack_cond (
        $_, $new_attrs->{where}
      );
    }
  }

  if (defined $old_having) {
    $new_attrs->{having} = $self->_stack_cond (
      $old_having, $new_attrs->{having}
    )
  }

  my $rs = (ref $self)->new($rsrc, $new_attrs);

  $rs->set_cache($cache) if ($cache);

  return $rs;
}

my $dark_sel_dumper;
sub _normalize_selection {
  my ($self, $attrs) = @_;

  # legacy syntax
  $attrs->{'+columns'} = $self->_merge_attr($attrs->{'+columns'}, delete $attrs->{include_columns})
    if exists $attrs->{include_columns};

  # columns are always placed first, however 

  # Keep the X vs +X separation until _resolved_attrs time - this allows to
  # delay the decision on whether to use a default select list ($rsrc->columns)
  # allowing stuff like the remove_columns helper to work
  #
  # select/as +select/+as pairs need special handling - the amount of select/as
  # elements in each pair does *not* have to be equal (think multicolumn
  # selectors like distinct(foo, bar) ). If the selector is bare (no 'as'
  # supplied at all) - try to infer the alias, either from the -as parameter
  # of the selector spec, or use the parameter whole if it looks like a column
  # name (ugly legacy heuristic). If all fails - leave the selector bare (which
  # is ok as well), but make sure no more additions to the 'as' chain take place
  for my $pref ('', '+') {

    my ($sel, $as) = map {
      my $key = "${pref}${_}";

      my $val = [ ref $attrs->{$key} eq 'ARRAY'
        ? @{$attrs->{$key}}
        : $attrs->{$key} || ()
      ];
      delete $attrs->{$key};
      $val;
    } qw/select as/;

    if (! @$as and ! @$sel ) {
      next;
    }
    elsif (@$as and ! @$sel) {
      $self->throw_exception(
        "Unable to handle ${pref}as specification (@$as) without a corresponding ${pref}select"
      );
    }
    elsif( ! @$as ) {
      # no as part supplied at all - try to deduce (unless explicit end of named selection is declared)
      # if any @$as has been supplied we assume the user knows what (s)he is doing
      # and blindly keep stacking up pieces
      unless ($attrs->{_dark_selector}) {
        SELECTOR:
        for (@$sel) {
          if ( ref $_ eq 'HASH' and exists $_->{-as} ) {
            push @$as, $_->{-as};
          }
          # assume any plain no-space, no-parenthesis string to be a column spec
          # FIXME - this is retarded but is necessary to support shit like 'count(foo)'
          elsif ( ! ref $_ and $_ =~ /^ [^\s\(\)]+ $/x) {
            push @$as, $_;
          }
          # if all else fails - raise a flag that no more aliasing will be allowed
          else {
            $attrs->{_dark_selector} = {
              plus_stage => $pref,
              string => ($dark_sel_dumper ||= do {
                  require Data::Dumper::Concise;
                  Data::Dumper::Concise::DumperObject()->Indent(0);
                })->Values([$_])->Dump
              ,
            };
            last SELECTOR;
          }
        }
      }
    }
    elsif (@$as < @$sel) {
      $self->throw_exception(
        "Unable to handle an ${pref}as specification (@$as) with less elements than the corresponding ${pref}select"
      );
    }
    elsif ($pref and $attrs->{_dark_selector}) {
      $self->throw_exception(
        "Unable to process named '+select', resultset contains an unnamed selector $attrs->{_dark_selector}{string}"
      );
    }


    # merge result
    $attrs->{"${pref}select"} = $self->_merge_attr($attrs->{"${pref}select"}, $sel);
    $attrs->{"${pref}as"} = $self->_merge_attr($attrs->{"${pref}as"}, $as);
  }
}

sub _stack_cond {
  my ($self, $left, $right) = @_;
  if (defined $left xor defined $right) {
    return defined $left ? $left : $right;
  }
  elsif (defined $left) {
    return { -and => [ map
      { ref $_ eq 'ARRAY' ? [ -or => $_ ] : $_ }
      ($left, $right)
    ]};
  }

  return undef;
}

sub search_literal {
  my ($self, $sql, @bind) = @_;
  my $attr;
  if ( @bind && ref($bind[-1]) eq 'HASH' ) {
    $attr = pop @bind;
  }
  return $self->search(\[ $sql, map [ __DUMMY__ => $_ ], @bind ], ($attr || () ));
}

sub find {
  my $self = shift;
  my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});

  my $rsrc = $self->result_source;

  # Parse out the condition from input
  my $call_cond;
  if (ref $_[0] eq 'HASH') {
    $call_cond = { %{$_[0]} };
  }
  else {
    my $constraint = exists $attrs->{key} ? $attrs->{key} : 'primary';
    my @c_cols = $rsrc->unique_constraint_columns($constraint);

    $self->throw_exception(
      "No constraint columns, maybe a malformed '$constraint' constraint?"
    ) unless @c_cols;

    $self->throw_exception (
      'find() expects either a column/value hashref, or a list of values '
    . "corresponding to the columns of the specified unique constraint '$constraint'"
    ) unless @c_cols == @_;

    $call_cond = {};
    @{$call_cond}{@c_cols} = @_;
  }

  my %related;
  for my $key (keys %$call_cond) {
    if (
      my $keyref = ref($call_cond->{$key})
        and
      my $relinfo = $rsrc->relationship_info($key)
    ) {
      my $val = delete $call_cond->{$key};

      next if $keyref eq 'ARRAY'; # has_many for multi_create

      my $rel_q = $rsrc->_resolve_condition(
        $relinfo->{cond}, $val, $key, $key
      );
      die "Can't handle complex relationship conditions in find" if ref($rel_q) ne 'HASH';
      @related{keys %$rel_q} = values %$rel_q;
    }
  }

  # relationship conditions take precedence (?)
  @{$call_cond}{keys %related} = values %related;

  my $alias = exists $attrs->{alias} ? $attrs->{alias} : $self->{attrs}{alias};
  my $final_cond;
  if (exists $attrs->{key}) {
    $final_cond = $self->_qualify_cond_columns (

      $self->_build_unique_cond (
        $attrs->{key},
        $call_cond,
      ),

      $alias,
    );
  }
  elsif ($self->{attrs}{accessor} and $self->{attrs}{accessor} eq 'single') {
    # This means that we got here after a merger of relationship conditions
    # in ::Relationship::Base::search_related (the row method), and furthermore
    # the relationship is of the 'single' type. This means that the condition
    # provided by the relationship (already attached to $self) is sufficient,
    # as there can be only one row in the database that would satisfy the
    # relationship
  }
  else {
    # no key was specified - fall down to heuristics mode:
    # run through all unique queries registered on the resultset, and
    # 'OR' all qualifying queries together
    my (@unique_queries, %seen_column_combinations);
    for my $c_name ($rsrc->unique_constraint_names) {
      next if $seen_column_combinations{
        join "\x00", sort $rsrc->unique_constraint_columns($c_name)
      }++;

      push @unique_queries, try {
        $self->_build_unique_cond ($c_name, $call_cond, 'croak_on_nulls')
      } || ();
    }

    $final_cond = @unique_queries
      ? [ map { $self->_qualify_cond_columns($_, $alias) } @unique_queries ]
      : $self->_non_unique_find_fallback ($call_cond, $attrs)
    ;
  }

  # Run the query, passing the result_class since it should propagate for find
  my $rs = $self->search ($final_cond, {result_class => $self->result_class, %$attrs});
  if (keys %{$rs->_resolved_attrs->{collapse}}) {
    my $row = $rs->next;
    carp "Query returned more than one row" if $rs->next;
    return $row;
  }
  else {
    return $rs->single;
  }
}

# This is a stop-gap method as agreed during the discussion on find() cleanup:
# http://lists.scsys.co.uk/pipermail/dbix-class/2010-October/009535.html
#
# It is invoked when find() is called in legacy-mode with insufficiently-unique
# condition. It is provided for overrides until a saner way forward is devised
#
# *NOTE* This is not a public method, and it's *GUARANTEED* to disappear down
# the road. Please adjust your tests accordingly to catch this situation early
# DBIx::Class::ResultSet->can('_non_unique_find_fallback') is reasonable
#
# The method will not be removed without an adequately complete replacement
# for strict-mode enforcement
sub _non_unique_find_fallback {
  my ($self, $cond, $attrs) = @_;

  return $self->_qualify_cond_columns(
    $cond,
    exists $attrs->{alias}
      ? $attrs->{alias}
      : $self->{attrs}{alias}
  );
}


sub _qualify_cond_columns {
  my ($self, $cond, $alias) = @_;

  my %aliased = %$cond;
  for (keys %aliased) {
    $aliased{"$alias.$_"} = delete $aliased{$_}
      if $_ !~ /\./;
  }

  return \%aliased;
}

sub _build_unique_cond {
  my ($self, $constraint_name, $extra_cond, $croak_on_null) = @_;

  my @c_cols = $self->result_source->unique_constraint_columns($constraint_name);

  # combination may fail if $self->{cond} is non-trivial
  my ($final_cond) = try {
    $self->_merge_with_rscond ($extra_cond)
  } catch {
    +{ %$extra_cond }
  };

  # trim out everything not in $columns
  $final_cond = { map {
    exists $final_cond->{$_}
      ? ( $_ => $final_cond->{$_} )
      : ()
  } @c_cols };

  if (my @missing = grep
    { ! ($croak_on_null ? defined $final_cond->{$_} : exists $final_cond->{$_}) }
    (@c_cols)
  ) {
    $self->throw_exception( sprintf ( "Unable to satisfy requested constraint '%s', no values for column(s): %s",
      $constraint_name,
      join (', ', map { "'$_'" } @missing),
    ) );
  }

  if (
    !$croak_on_null
      and
    !$ENV{DBIC_NULLABLE_KEY_NOWARN}
      and
    my @undefs = grep { ! defined $final_cond->{$_} } (keys %$final_cond)
  ) {
    carp_unique ( sprintf (
      "NULL/undef values supplied for requested unique constraint '%s' (NULL "
    . 'values in column(s): %s). This is almost certainly not what you wanted, '
    . 'though you can set DBIC_NULLABLE_KEY_NOWARN to disable this warning.',
      $constraint_name,
      join (', ', map { "'$_'" } @undefs),
    ));
  }

  return $final_cond;
}

sub search_related {
  return shift->related_resultset(shift)->search(@_);
}

sub search_related_rs {
  return shift->related_resultset(shift)->search_rs(@_);
}

sub cursor {
  my ($self) = @_;

  my $attrs = $self->_resolved_attrs_copy;

  return $self->{cursor}
    ||= $self->result_source->storage->select($attrs->{from}, $attrs->{select},
          $attrs->{where},$attrs);
}

sub single {
  my ($self, $where) = @_;
  if(@_ > 2) {
      $self->throw_exception('single() only takes search conditions, no attributes. You want ->search( $cond, $attrs )->single()');
  }

  my $attrs = $self->_resolved_attrs_copy;

  if (keys %{$attrs->{collapse}}) {
    $self->throw_exception(
      'single() can not be used on resultsets prefetching has_many. Use find( \%cond ) or next() instead'
    );
  }

  if ($where) {
    if (defined $attrs->{where}) {
      $attrs->{where} = {
        '-and' =>
            [ map { ref $_ eq 'ARRAY' ? [ -or => $_ ] : $_ }
               $where, delete $attrs->{where} ]
      };
    } else {
      $attrs->{where} = $where;
    }
  }

  my @data = $self->result_source->storage->select_single(
    $attrs->{from}, $attrs->{select},
    $attrs->{where}, $attrs
  );

  return (@data ? ($self->_construct_object(@data))[0] : undef);
}


# _collapse_query
#
# Recursively collapse the query, accumulating values for each column.

sub _collapse_query {
  my ($self, $query, $collapsed) = @_;

  $collapsed ||= {};

  if (ref $query eq 'ARRAY') {
    foreach my $subquery (@$query) {
      next unless ref $subquery;  # -or
      $collapsed = $self->_collapse_query($subquery, $collapsed);
    }
  }
  elsif (ref $query eq 'HASH') {
    if (keys %$query and (keys %$query)[0] eq '-and') {
      foreach my $subquery (@{$query->{-and}}) {
        $collapsed = $self->_collapse_query($subquery, $collapsed);
      }
    }
    else {
      foreach my $col (keys %$query) {
        my $value = $query->{$col};
        $collapsed->{$col}{$value}++;
      }
    }
  }

  return $collapsed;
}

sub get_column {
  my ($self, $column) = @_;
  my $new = DBIx::Class::ResultSetColumn->new($self, $column);
  return $new;
}

sub search_like {
  my $class = shift;
  carp_unique (
    'search_like() is deprecated and will be removed in DBIC version 0.09.'
   .' Instead use ->search({ x => { -like => "y%" } })'
   .' (note the outer pair of {}s - they are important!)'
  );
  my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
  my $query = ref $_[0] eq 'HASH' ? { %{shift()} }: {@_};
  $query->{$_} = { 'like' => $query->{$_} } for keys %$query;
  return $class->search($query, { %$attrs });
}

sub slice {
  my ($self, $min, $max) = @_;
  my $attrs = {}; # = { %{ $self->{attrs} || {} } };
  $attrs->{offset} = $self->{attrs}{offset} || 0;
  $attrs->{offset} += $min;
  $attrs->{rows} = ($max ? ($max - $min + 1) : 1);
  return $self->search(undef, $attrs);
  #my $slice = (ref $self)->new($self->result_source, $attrs);
  #return (wantarray ? $slice->all : $slice);
}

sub next {
  my ($self) = @_;
  if (my $cache = $self->get_cache) {
    $self->{all_cache_position} ||= 0;
    return $cache->[$self->{all_cache_position}++];
  }
  if ($self->{attrs}{cache}) {
    delete $self->{pager};
    $self->{all_cache_position} = 1;
    return ($self->all)[0];
  }
  if ($self->{stashed_objects}) {
    my $obj = shift(@{$self->{stashed_objects}});
    delete $self->{stashed_objects} unless @{$self->{stashed_objects}};
    return $obj;
  }
  my @row = (
    exists $self->{stashed_row}
      ? @{delete $self->{stashed_row}}
      : $self->cursor->next
  );
  return undef unless (@row);
  my ($row, @more) = $self->_construct_object(@row);
  $self->{stashed_objects} = \@more if @more;
  return $row;
}

sub _construct_object {
  my ($self, @row) = @_;

  my $info = $self->_collapse_result($self->{_attrs}{as}, \@row)
    or return ();
  my @new = $self->result_class->inflate_result($self->result_source, @$info);
  @new = $self->{_attrs}{record_filter}->(@new)
    if exists $self->{_attrs}{record_filter};
  return @new;
}

sub _collapse_result {
  my ($self, $as_proto, $row) = @_;

  my @copy = @$row;

  # 'foo'         => [ undef, 'foo' ]
  # 'foo.bar'     => [ 'foo', 'bar' ]
  # 'foo.bar.baz' => [ 'foo.bar', 'baz' ]

  my @construct_as = map { [ (/^(?:(.*)\.)?([^.]+)$/) ] } @$as_proto;

  my %collapse = %{$self->{_attrs}{collapse}||{}};

  my @pri_index;

  # if we're doing collapsing (has_many prefetch) we need to grab records
  # until the PK changes, so fill @pri_index. if not, we leave it empty so
  # we know we don't have to bother.

  # the reason for not using the collapse stuff directly is because if you
  # had for e.g. two artists in a row with no cds, the collapse info for
  # both would be NULL (undef) so you'd lose the second artist

  # store just the index so we can check the array positions from the row
  # without having to contruct the full hash

  if (keys %collapse) {
    my %pri = map { ($_ => 1) } $self->result_source->_pri_cols;
    foreach my $i (0 .. $#construct_as) {
      next if defined($construct_as[$i][0]); # only self table
      if (delete $pri{$construct_as[$i][1]}) {
        push(@pri_index, $i);
      }
      last unless keys %pri; # short circuit (Johnny Five Is Alive!)
    }
  }

  # no need to do an if, it'll be empty if @pri_index is empty anyway

  my %pri_vals = map { ($_ => $copy[$_]) } @pri_index;

  my @const_rows;

  do { # no need to check anything at the front, we always want the first row

    my %const;

    foreach my $this_as (@construct_as) {
      $const{$this_as->[0]||''}{$this_as->[1]} = shift(@copy);
    }

    push(@const_rows, \%const);

  } until ( # no pri_index => no collapse => drop straight out
      !@pri_index
    or
      do { # get another row, stash it, drop out if different PK

        @copy = $self->cursor->next;
        $self->{stashed_row} = \@copy;

        # last thing in do block, counts as true if anything doesn't match

        # check xor defined first for NULL vs. NOT NULL then if one is
        # defined the other must be so check string equality

        grep {
          (defined $pri_vals{$_} ^ defined $copy[$_])
          || (defined $pri_vals{$_} && ($pri_vals{$_} ne $copy[$_]))
        } @pri_index;
      }
  );

  my $alias = $self->{attrs}{alias};
  my $info = [];

  my %collapse_pos;

  my @const_keys;

  foreach my $const (@const_rows) {
    scalar @const_keys or do {
      @const_keys = sort { length($a) <=> length($b) } keys %$const;
    };
    foreach my $key (@const_keys) {
      if (length $key) {
        my $target = $info;
        my @parts = split(/\./, $key);
        my $cur = '';
        my $data = $const->{$key};
        foreach my $p (@parts) {
          $target = $target->[1]->{$p} ||= [];
          $cur .= ".${p}";
          if ($cur eq ".${key}" && (my @ckey = @{$collapse{$cur}||[]})) {
            # collapsing at this point and on final part
            my $pos = $collapse_pos{$cur};
            CK: foreach my $ck (@ckey) {
              if (!defined $pos->{$ck} || $pos->{$ck} ne $data->{$ck}) {
                $collapse_pos{$cur} = $data;
                delete @collapse_pos{ # clear all positioning for sub-entries
                  grep { m/^\Q${cur}.\E/ } keys %collapse_pos
                };
                push(@$target, []);
                last CK;
              }
            }
          }
          if (exists $collapse{$cur}) {
            $target = $target->[-1];
          }
        }
        $target->[0] = $data;
      } else {
        $info->[0] = $const->{$key};
      }
    }
  }

  return $info;
}

sub result_class {
  my ($self, $result_class) = @_;
  if ($result_class) {
    unless (ref $result_class) { # don't fire this for an object
      $self->ensure_class_loaded($result_class);
    }
    $self->_result_class($result_class);
    # THIS LINE WOULD BE A BUG - this accessor specifically exists to
    # permit the user to set result class on one result set only; it only
    # chains if provided to search()
    #$self->{attrs}{result_class} = $result_class if ref $self;
  }
  $self->_result_class;
}

sub count {
  my $self = shift;
  return $self->search(@_)->count if @_ and defined $_[0];
  return scalar @{ $self->get_cache } if $self->get_cache;

  my $attrs = $self->_resolved_attrs_copy;

  # this is a little optimization - it is faster to do the limit
  # adjustments in software, instead of a subquery
  my $rows = delete $attrs->{rows};
  my $offset = delete $attrs->{offset};

  my $crs;
  if ($self->_has_resolved_attr (qw/collapse group_by/)) {
    $crs = $self->_count_subq_rs ($attrs);
  }
  else {
    $crs = $self->_count_rs ($attrs);
  }
  my $count = $crs->next;

  $count -= $offset if $offset;
  $count = $rows if $rows and $rows < $count;
  $count = 0 if ($count < 0);

  return $count;
}

sub count_rs {
  my $self = shift;
  return $self->search(@_)->count_rs if @_;

  # this may look like a lack of abstraction (count() does about the same)
  # but in fact an _rs *must* use a subquery for the limits, as the
  # software based limiting can not be ported if this $rs is to be used
  # in a subquery itself (i.e. ->as_query)
  if ($self->_has_resolved_attr (qw/collapse group_by offset rows/)) {
    return $self->_count_subq_rs;
  }
  else {
    return $self->_count_rs;
  }
}

#
# returns a ResultSetColumn object tied to the count query
#
sub _count_rs {
  my ($self, $attrs) = @_;

  my $rsrc = $self->result_source;
  $attrs ||= $self->_resolved_attrs;

  my $tmp_attrs = { %$attrs };
  # take off any limits, record_filter is cdbi, and no point of ordering nor locking a count
  delete @{$tmp_attrs}{qw/rows offset order_by record_filter for/};

  # overwrite the selector (supplied by the storage)
  $tmp_attrs->{select} = $rsrc->storage->_count_select ($rsrc, $attrs);
  $tmp_attrs->{as} = 'count';
  delete @{$tmp_attrs}{qw/columns/};

  my $tmp_rs = $rsrc->resultset_class->new($rsrc, $tmp_attrs)->get_column ('count');

  return $tmp_rs;
}

#
# same as above but uses a subquery
#
sub _count_subq_rs {
  my ($self, $attrs) = @_;

  my $rsrc = $self->result_source;
  $attrs ||= $self->_resolved_attrs;

  my $sub_attrs = { %$attrs };
  # extra selectors do not go in the subquery and there is no point of ordering it, nor locking it
  delete @{$sub_attrs}{qw/collapse columns as select _prefetch_selector_range order_by for/};

  # if we multi-prefetch we group_by primary keys only as this is what we would
  # get out of the rs via ->next/->all. We *DO WANT* to clobber old group_by regardless
  if ( keys %{$attrs->{collapse}}  ) {
    $sub_attrs->{group_by} = [ map { "$attrs->{alias}.$_" } ($rsrc->_pri_cols) ]
  }

  # Calculate subquery selector
  if (my $g = $sub_attrs->{group_by}) {

    my $sql_maker = $rsrc->storage->sql_maker;

    # necessary as the group_by may refer to aliased functions
    my $sel_index;
    for my $sel (@{$attrs->{select}}) {
      $sel_index->{$sel->{-as}} = $sel
        if (ref $sel eq 'HASH' and $sel->{-as});
    }

    # anything from the original select mentioned on the group-by needs to make it to the inner selector
    # also look for named aggregates referred in the having clause
    # having often contains scalarrefs - thus parse it out entirely
    my @parts = @$g;
    if ($attrs->{having}) {
      local $sql_maker->{having_bind};
      local $sql_maker->{quote_char} = $sql_maker->{quote_char};
      local $sql_maker->{name_sep} = $sql_maker->{name_sep};
      unless (defined $sql_maker->{quote_char} and length $sql_maker->{quote_char}) {
        $sql_maker->{quote_char} = [ "\x00", "\xFF" ];
        # if we don't unset it we screw up retarded but unfortunately working
        # 'MAX(foo.bar)' => { '>', 3 }
        $sql_maker->{name_sep} = '';
      }

      my ($lquote, $rquote, $sep) = map { quotemeta $_ } ($sql_maker->_quote_chars, $sql_maker->name_sep);

      my $sql = $sql_maker->_parse_rs_attrs ({ having => $attrs->{having} });

      # search for both a proper quoted qualified string, for a naive unquoted scalarref
      # and if all fails for an utterly naive quoted scalar-with-function
      while ($sql =~ /
                $rquote $sep $lquote (.+?) $rquote
                    |
                [\s,] \w+ \. (\w+) [\s,]
                    |
                [\s,] $lquote (.+?) $rquote [\s,]
            /gx) {
        push @parts, ($1 || $2 || $3);  # one of them matched if we got here
      }
    }

    for (@parts) {
      my $colpiece = $sel_index->{$_} || $_;

      # unqualify join-based group_by's. Arcane but possible query
      # also horrible horrible hack to alias a column (not a func.)
      # (probably need to introduce SQLA syntax)
      if ($colpiece =~ /\./ && $colpiece !~ /^$attrs->{alias}\./) {
        my $as = $colpiece;
        $as =~ s/\./__/;
        $colpiece = \ sprintf ('%s AS %s', map { $sql_maker->_quote ($_) } ($colpiece, $as) );
      }
      push @{$sub_attrs->{select}}, $colpiece;
    }
  }
  else {
    my @pcols = map { "$attrs->{alias}.$_" } ($rsrc->primary_columns);
    $sub_attrs->{select} = @pcols ? \@pcols : [ 1 ];
  }

  return $rsrc->resultset_class
               ->new ($rsrc, $sub_attrs)
                ->as_subselect_rs
                 ->search ({}, { columns => { count => $rsrc->storage->_count_select ($rsrc, $attrs) } })
                  ->get_column ('count');
}

sub _bool {
  return 1;
}

sub count_literal { shift->search_literal(@_)->count; }

sub all {
  my $self = shift;
  if(@_) {
      $self->throw_exception("all() doesn't take any arguments, you probably wanted ->search(...)->all()");
  }

  return @{ $self->get_cache } if $self->get_cache;

  my @obj;

  if (keys %{$self->_resolved_attrs->{collapse}}) {
    # Using $self->cursor->all is really just an optimisation.
    # If we're collapsing has_many prefetches it probably makes
    # very little difference, and this is cleaner than hacking
    # _construct_object to survive the approach
    $self->cursor->reset;
    my @row = $self->cursor->next;
    while (@row) {
      push(@obj, $self->_construct_object(@row));
      @row = (exists $self->{stashed_row}
               ? @{delete $self->{stashed_row}}
               : $self->cursor->next);
    }
  } else {
    @obj = map { $self->_construct_object(@$_) } $self->cursor->all;
  }

  $self->set_cache(\@obj) if $self->{attrs}{cache};

  return @obj;
}

sub reset {
  my ($self) = @_;
  delete $self->{_attrs} if exists $self->{_attrs};
  $self->{all_cache_position} = 0;
  $self->cursor->reset;
  return $self;
}

sub first {
  return $_[0]->reset->next;
}


# _rs_update_delete
#
# Determines whether and what type of subquery is required for the $rs operation.
# If grouping is necessary either supplies its own, or verifies the current one
# After all is done delegates to the proper storage method.

sub _rs_update_delete {
  my ($self, $op, $values) = @_;

  my $rsrc = $self->result_source;

  my $needs_group_by_subq = $self->_has_resolved_attr (qw/collapse group_by -join/);
  my $needs_subq = $needs_group_by_subq || $self->_has_resolved_attr(qw/rows offset/);

  if ($needs_group_by_subq or $needs_subq) {

    # make a new $rs selecting only the PKs (that's all we really need)
    my $attrs = $self->_resolved_attrs_copy;


    delete $attrs->{$_} for qw/collapse _collapse_order_by select _prefetch_selector_range as/;
    $attrs->{columns} = [ map { "$attrs->{alias}.$_" } ($self->result_source->_pri_cols) ];

    if ($needs_group_by_subq) {
      # make sure no group_by was supplied, or if there is one - make sure it matches
      # the columns compiled above perfectly. Anything else can not be sanely executed
      # on most databases so croak right then and there

      if (my $g = $attrs->{group_by}) {
        my @current_group_by = map
          { $_ =~ /\./ ? $_ : "$attrs->{alias}.$_" }
          @$g
        ;

        if (
          join ("\x00", sort @current_group_by)
            ne
          join ("\x00", sort @{$attrs->{columns}} )
        ) {
          $self->throw_exception (
            "You have just attempted a $op operation on a resultset which does group_by"
            . ' on columns other than the primary keys, while DBIC internally needs to retrieve'
            . ' the primary keys in a subselect. All sane RDBMS engines do not support this'
            . ' kind of queries. Please retry the operation with a modified group_by or'
            . ' without using one at all.'
          );
        }
      }
      else {
        $attrs->{group_by} = $attrs->{columns};
      }
    }

    my $subrs = (ref $self)->new($rsrc, $attrs);
    return $self->result_source->storage->_subq_update_delete($subrs, $op, $values);
  }
  else {
    # Most databases do not allow aliasing of tables in UPDATE/DELETE. Thus
    # a condition containing 'me' or other table prefixes will not work
    # at all. What this code tries to do (badly) is to generate a condition
    # with the qualifiers removed, by exploiting the quote mechanism of sqla
    #
    # this is atrocious and should be replaced by normal sqla introspection
    # one sunny day
    my ($sql, @bind) = do {
      my $sqla = $rsrc->storage->sql_maker;
      local $sqla->{_dequalify_idents} = 1;
      $sqla->_recurse_where($self->{cond});
    } if $self->{cond};

    return $rsrc->storage->$op(
      $rsrc,
      $op eq 'update' ? $values : (),
      $self->{cond} ? \[$sql, @bind] : (),
    );
  }
}

sub update {
  my ($self, $values) = @_;
  $self->throw_exception('Values for update must be a hash')
    unless ref $values eq 'HASH';

  return $self->_rs_update_delete ('update', $values);
}

sub update_all {
  my ($self, $values) = @_;
  $self->throw_exception('Values for update_all must be a hash')
    unless ref $values eq 'HASH';

  my $guard = $self->result_source->schema->txn_scope_guard;
  $_->update($values) for $self->all;
  $guard->commit;
  return 1;
}

sub delete {
  my $self = shift;
  $self->throw_exception('delete does not accept any arguments')
    if @_;

  return $self->_rs_update_delete ('delete');
}

sub delete_all {
  my $self = shift;
  $self->throw_exception('delete_all does not accept any arguments')
    if @_;

  my $guard = $self->result_source->schema->txn_scope_guard;
  $_->delete for $self->all;
  $guard->commit;
  return 1;
}

sub populate {
  my $self = shift;

  # cruft placed in standalone method
  my $data = $self->_normalize_populate_args(@_);

  if(defined wantarray) {
    my @created;
    foreach my $item (@$data) {
      push(@created, $self->create($item));
    }
    return wantarray ? @created : \@created;
  } 
  else {
    my $first = $data->[0];

    # if a column is a registered relationship, and is a non-blessed hash/array, consider
    # it relationship data
    my (@rels, @columns);
    my $rsrc = $self->result_source;
    my $rels = { map { $_ => $rsrc->relationship_info($_) } $rsrc->relationships };
    for (keys %$first) {
      my $ref = ref $first->{$_};
      $rels->{$_} && ($ref eq 'ARRAY' or $ref eq 'HASH')
        ? push @rels, $_
        : push @columns, $_
      ;
    }

    my @pks = $rsrc->primary_columns;

    ## do the belongs_to relationships
    foreach my $index (0..$#$data) {

      # delegate to create() for any dataset without primary keys with specified relationships
      if (grep { !defined $data->[$index]->{$_} } @pks ) {
        for my $r (@rels) {
          if (grep { ref $data->[$index]{$r} eq $_ } qw/HASH ARRAY/) {  # a related set must be a HASH or AoH
            my @ret = $self->populate($data);
            return;
          }
        }
      }

      foreach my $rel (@rels) {
        next unless ref $data->[$index]->{$rel} eq "HASH";
        my $result = $self->related_resultset($rel)->create($data->[$index]->{$rel});
        my ($reverse_relname, $reverse_relinfo) = %{$rsrc->reverse_relationship_info($rel)};
        my $related = $result->result_source->_resolve_condition(
          $reverse_relinfo->{cond},
          $self,
          $result,
          $rel,
        );

        delete $data->[$index]->{$rel};
        $data->[$index] = {%{$data->[$index]}, %$related};

        push @columns, keys %$related if $index == 0;
      }
    }

    ## inherit the data locked in the conditions of the resultset
    my ($rs_data) = $self->_merge_with_rscond({});
    delete @{$rs_data}{@columns};
    my @inherit_cols = keys %$rs_data;
    my @inherit_data = values %$rs_data;

    ## do bulk insert on current row
    $rsrc->storage->insert_bulk(
      $rsrc,
      [@columns, @inherit_cols],
      [ map { [ @$_{@columns}, @inherit_data ] } @$data ],
    );

    ## do the has_many relationships
    foreach my $item (@$data) {

      my $main_row;

      foreach my $rel (@rels) {
        next unless ref $item->{$rel} eq "ARRAY" && @{ $item->{$rel} };

        $main_row ||= $self->new_result({map { $_ => $item->{$_} } @pks});

        my $child = $main_row->$rel;

        my $related = $child->result_source->_resolve_condition(
          $rels->{$rel}{cond},
          $child,
          $main_row,
          $rel,
        );

        my @rows_to_add = ref $item->{$rel} eq 'ARRAY' ? @{$item->{$rel}} : ($item->{$rel});
        my @populate = map { {%$_, %$related} } @rows_to_add;

        $child->populate( \@populate );
      }
    }
  }
}


# populate() argumnets went over several incarnations
# What we ultimately support is AoH
sub _normalize_populate_args {
  my ($self, $arg) = @_;

  if (ref $arg eq 'ARRAY') {
    if (ref $arg->[0] eq 'HASH') {
      return $arg;
    }
    elsif (ref $arg->[0] eq 'ARRAY') {
      my @ret;
      my @colnames = @{$arg->[0]};
      foreach my $values (@{$arg}[1 .. $#$arg]) {
        push @ret, { map { $colnames[$_] => $values->[$_] } (0 .. $#colnames) };
      }
      return \@ret;
    }
  }

  $self->throw_exception('Populate expects an arrayref of hashrefs or arrayref of arrayrefs');
}

# make a wizard good for both a scalar and a hashref
my $mk_lazy_count_wizard = sub {
  require Variable::Magic;

  my $stash = { total_rs => shift };
  my $slot = shift; # only used by the hashref magic

  my $magic = Variable::Magic::wizard (
    data => sub { $stash },

    (!$slot)
    ? (
      # the scalar magic
      get => sub {
        # set value lazily, and dispell for good
        ${$_[0]} = $_[1]{total_rs}->count;
        Variable::Magic::dispell (${$_[0]}, $_[1]{magic_selfref});
        return 1;
      },
      set => sub {
        # an explicit set implies dispell as well
        # the unless() is to work around "fun and giggles" below
        Variable::Magic::dispell (${$_[0]}, $_[1]{magic_selfref})
          unless (caller(2))[3] eq 'DBIx::Class::ResultSet::pager';
        return 1;
      },
    )
    : (
      # the uvar magic
      fetch => sub {
        if ($_[2] eq $slot and !$_[1]{inactive}) {
          my $cnt = $_[1]{total_rs}->count;
          $_[0]->{$slot} = $cnt;

          # attempting to dispell in a fetch handle (works in store), seems
          # to invariable segfault on 5.10, 5.12, 5.13 :(
          # so use an inactivator instead
          #Variable::Magic::dispell (%{$_[0]}, $_[1]{magic_selfref});
          $_[1]{inactive}++;
        }
        return 1;
      },
      store => sub {
        if (! $_[1]{inactive} and $_[2] eq $slot) {
          #Variable::Magic::dispell (%{$_[0]}, $_[1]{magic_selfref});
          $_[1]{inactive}++
            unless (caller(2))[3] eq 'DBIx::Class::ResultSet::pager';
        }
        return 1;
      },
    ),
  );

  $stash->{magic_selfref} = $magic;
  weaken ($stash->{magic_selfref}); # this fails on 5.8.1

  return $magic;
};

# the tie class for 5.8.1
{
  package # hide from pause
    DBIx::Class::__DBIC_LAZY_RS_COUNT__;
  use base qw/Tie::Hash/;

  sub FIRSTKEY { my $dummy = scalar keys %{$_[0]{data}}; each %{$_[0]{data}} }
  sub NEXTKEY  { each %{$_[0]{data}} }
  sub EXISTS   { exists $_[0]{data}{$_[1]} }
  sub DELETE   { delete $_[0]{data}{$_[1]} }
  sub CLEAR    { %{$_[0]{data}} = () }
  sub SCALAR   { scalar %{$_[0]{data}} }

  sub TIEHASH {
    $_[1]{data} = {%{$_[1]{selfref}}};
    %{$_[1]{selfref}} = ();
    Scalar::Util::weaken ($_[1]{selfref});
    return bless ($_[1], $_[0]);
  };

  sub FETCH {
    if ($_[1] eq $_[0]{slot}) {
      my $cnt = $_[0]{data}{$_[1]} = $_[0]{total_rs}->count;
      untie %{$_[0]{selfref}};
      %{$_[0]{selfref}} = %{$_[0]{data}};
      return $cnt;
    }
    else {
      $_[0]{data}{$_[1]};
    }
  }

  sub STORE {
    $_[0]{data}{$_[1]} = $_[2];
    if ($_[1] eq $_[0]{slot}) {
      untie %{$_[0]{selfref}};
      %{$_[0]{selfref}} = %{$_[0]{data}};
    }
    $_[2];
  }
}

sub pager {
  my ($self) = @_;

  return $self->{pager} if $self->{pager};

  my $attrs = $self->{attrs};
  if (!defined $attrs->{page}) {
    $self->throw_exception("Can't create pager for non-paged rs");
  }
  elsif ($attrs->{page} <= 0) {
    $self->throw_exception('Invalid page number (page-numbers are 1-based)');
  }
  $attrs->{rows} ||= 10;

  # throw away the paging flags and re-run the count (possibly
  # with a subselect) to get the real total count
  my $count_attrs = { %$attrs };
  delete $count_attrs->{$_} for qw/rows offset page pager/;
  my $total_rs = (ref $self)->new($self->result_source, $count_attrs);


### the following may seem awkward and dirty, but it's a thought-experiment
### necessary for future development of DBIx::DS. Do *NOT* change this code
### before talking to ribasushi/mst

  require Data::Page;
  my $pager = Data::Page->new(
    0,  #start with an empty set
    $attrs->{rows},
    $self->{attrs}{page},
  );

  my $data_slot = 'total_entries';

  # Since we are interested in a cached value (once it's set - it's set), every
  # technique will detach from the magic-host once the time comes to fire the
  # ->count (or in the segfaulting case of >= 5.10 it will deactivate itself)

  if ($] < 5.008003) {
    # 5.8.1 throws 'Modification of a read-only value attempted' when one tries
    # to weakref the magic container :(
    # tested on 5.8.1
    tie (%$pager, 'DBIx::Class::__DBIC_LAZY_RS_COUNT__',
      { slot => $data_slot, total_rs => $total_rs, selfref => $pager }
    );
  }
  elsif ($] < 5.010) {
    # We can use magic on the hash value slot. It's interesting that the magic is
    # attached to the hash-slot, and does *not* stop working once I do the dummy
    # assignments after the cast()
    # tested on 5.8.3 and 5.8.9
    my $magic = $mk_lazy_count_wizard->($total_rs);
    Variable::Magic::cast ( $pager->{$data_slot}, $magic );

    # this is for fun and giggles
    $pager->{$data_slot} = -1;
    $pager->{$data_slot} = 0;

    # this does not work for scalars, but works with
    # uvar magic below
    #my %vals = %$pager;
    #%$pager = ();
    #%{$pager} = %vals;
  }
  else {
    # And the uvar magic
    # works on 5.10.1, 5.12.1 and 5.13.4 in its current form,
    # however see the wizard maker for more notes
    my $magic = $mk_lazy_count_wizard->($total_rs, $data_slot);
    Variable::Magic::cast ( %$pager, $magic );

    # still works
    $pager->{$data_slot} = -1;
    $pager->{$data_slot} = 0;

    # this now works
    my %vals = %$pager;
    %$pager = ();
    %{$pager} = %vals;
  }

  return $self->{pager} = $pager;
}

sub page {
  my ($self, $page) = @_;
  return (ref $self)->new($self->result_source, { %{$self->{attrs}}, page => $page });
}

sub new_result {
  my ($self, $values) = @_;
  $self->throw_exception( "new_result needs a hash" )
    unless (ref $values eq 'HASH');

  my ($merged_cond, $cols_from_relations) = $self->_merge_with_rscond($values);

  my %new = (
    %$merged_cond,
    @$cols_from_relations
      ? (-cols_from_relations => $cols_from_relations)
      : (),
    -result_source => $self->result_source, # DO NOT REMOVE THIS, REQUIRED
  );

  return $self->result_class->new(\%new);
}

# _merge_with_rscond
#
# Takes a simple hash of K/V data and returns its copy merged with the
# condition already present on the resultset. Additionally returns an
# arrayref of value/condition names, which were inferred from related
# objects (this is needed for in-memory related objects)
sub _merge_with_rscond {
  my ($self, $data) = @_;

  my (%new_data, @cols_from_relations);

  my $alias = $self->{attrs}{alias};

  if (! defined $self->{cond}) {
    # just massage $data below
  }
  elsif ($self->{cond} eq $DBIx::Class::ResultSource::UNRESOLVABLE_CONDITION) {
    %new_data = %{ $self->{attrs}{related_objects} || {} };  # nothing might have been inserted yet
    @cols_from_relations = keys %new_data;
  }
  elsif (ref $self->{cond} ne 'HASH') {
    $self->throw_exception(
      "Can't abstract implicit construct, resultset condition not a hash"
    );
  }
  else {
    # precendence must be given to passed values over values inherited from
    # the cond, so the order here is important.
    my $collapsed_cond = $self->_collapse_cond($self->{cond});
    my %implied = %{$self->_remove_alias($collapsed_cond, $alias)};

    while ( my($col, $value) = each %implied ) {
      my $vref = ref $value;
      if (
        $vref eq 'HASH'
          and
        keys(%$value) == 1
          and
        (keys %$value)[0] eq '='
      ) {
        $new_data{$col} = $value->{'='};
      }
      elsif( !$vref or $vref eq 'SCALAR' or blessed($value) ) {
        $new_data{$col} = $value;
      }
    }
  }

  %new_data = (
    %new_data,
    %{ $self->_remove_alias($data, $alias) },
  );

  return (\%new_data, \@cols_from_relations);
}

# _has_resolved_attr
#
# determines if the resultset defines at least one
# of the attributes supplied
#
# used to determine if a subquery is neccessary
#
# supports some virtual attributes:
#   -join
#     This will scan for any joins being present on the resultset.
#     It is not a mere key-search but a deep inspection of {from}
#

sub _has_resolved_attr {
  my ($self, @attr_names) = @_;

  my $attrs = $self->_resolved_attrs;

  my %extra_checks;

  for my $n (@attr_names) {
    if (grep { $n eq $_ } (qw/-join/) ) {
      $extra_checks{$n}++;
      next;
    }

    my $attr =  $attrs->{$n};

    next if not defined $attr;

    if (ref $attr eq 'HASH') {
      return 1 if keys %$attr;
    }
    elsif (ref $attr eq 'ARRAY') {
      return 1 if @$attr;
    }
    else {
      return 1 if $attr;
    }
  }

  # a resolved join is expressed as a multi-level from
  return 1 if (
    $extra_checks{-join}
      and
    ref $attrs->{from} eq 'ARRAY'
      and
    @{$attrs->{from}} > 1
  );

  return 0;
}

# _collapse_cond
#
# Recursively collapse the condition.

sub _collapse_cond {
  my ($self, $cond, $collapsed) = @_;

  $collapsed ||= {};

  if (ref $cond eq 'ARRAY') {
    foreach my $subcond (@$cond) {
      next unless ref $subcond;  # -or
      $collapsed = $self->_collapse_cond($subcond, $collapsed);
    }
  }
  elsif (ref $cond eq 'HASH') {
    if (keys %$cond and (keys %$cond)[0] eq '-and') {
      foreach my $subcond (@{$cond->{-and}}) {
        $collapsed = $self->_collapse_cond($subcond, $collapsed);
      }
    }
    else {
      foreach my $col (keys %$cond) {
        my $value = $cond->{$col};
        $collapsed->{$col} = $value;
      }
    }
  }

  return $collapsed;
}

# _remove_alias
#
# Remove the specified alias from the specified query hash. A copy is made so
# the original query is not modified.

sub _remove_alias {
  my ($self, $query, $alias) = @_;

  my %orig = %{ $query || {} };
  my %unaliased;

  foreach my $key (keys %orig) {
    if ($key !~ /\./) {
      $unaliased{$key} = $orig{$key};
      next;
    }
    $unaliased{$1} = $orig{$key}
      if $key =~ m/^(?:\Q$alias\E\.)?([^.]+)$/;
  }

  return \%unaliased;
}

sub as_query {
  my $self = shift;

  my $attrs = $self->_resolved_attrs_copy;

  # For future use:
  #
  # in list ctx:
  # my ($sql, \@bind, \%dbi_bind_attrs) = _select_args_to_query (...)
  # $sql also has no wrapping parenthesis in list ctx
  #
  my $sqlbind = $self->result_source->storage
    ->_select_args_to_query ($attrs->{from}, $attrs->{select}, $attrs->{where}, $attrs);

  return $sqlbind;
}

sub find_or_new {
  my $self     = shift;
  my $attrs    = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
  my $hash     = ref $_[0] eq 'HASH' ? shift : {@_};
  if (keys %$hash and my $row = $self->find($hash, $attrs) ) {
    return $row;
  }
  return $self->new_result($hash);
}

sub create {
  my ($self, $attrs) = @_;
  $self->throw_exception( "create needs a hashref" )
    unless ref $attrs eq 'HASH';
  return $self->new_result($attrs)->insert;
}

sub find_or_create {
  my $self     = shift;
  my $attrs    = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
  my $hash     = ref $_[0] eq 'HASH' ? shift : {@_};
  if (keys %$hash and my $row = $self->find($hash, $attrs) ) {
    return $row;
  }
  return $self->create($hash);
}

sub update_or_create {
  my $self = shift;
  my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
  my $cond = ref $_[0] eq 'HASH' ? shift : {@_};

  my $row = $self->find($cond, $attrs);
  if (defined $row) {
    $row->update($cond);
    return $row;
  }

  return $self->create($cond);
}

sub update_or_new {
    my $self  = shift;
    my $attrs = ( @_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {} );
    my $cond  = ref $_[0] eq 'HASH' ? shift : {@_};

    my $row = $self->find( $cond, $attrs );
    if ( defined $row ) {
        $row->update($cond);
        return $row;
    }

    return $self->new_result($cond);
}

sub get_cache {
  shift->{all_cache};
}

sub set_cache {
  my ( $self, $data ) = @_;
  $self->throw_exception("set_cache requires an arrayref")
      if defined($data) && (ref $data ne 'ARRAY');
  $self->{all_cache} = $data;
}

sub clear_cache {
  shift->set_cache(undef);
}

sub is_paged {
  my ($self) = @_;
  return !!$self->{attrs}{page};
}

sub is_ordered {
  my ($self) = @_;
  return scalar $self->result_source->storage->_extract_order_criteria($self->{attrs}{order_by});
}

sub related_resultset {
  my ($self, $rel) = @_;

  $self->{related_resultsets} ||= {};
  return $self->{related_resultsets}{$rel} ||= do {
    my $rsrc = $self->result_source;
    my $rel_info = $rsrc->relationship_info($rel);

    $self->throw_exception(
      "search_related: result source '" . $rsrc->source_name .
        "' has no such relationship $rel")
      unless $rel_info;

    my $attrs = $self->_chain_relationship($rel);

    my $join_count = $attrs->{seen_join}{$rel};

    my $alias = $self->result_source->storage
        ->relname_to_table_alias($rel, $join_count);

    # since this is search_related, and we already slid the select window inwards
    # (the select/as attrs were deleted in the beginning), we need to flip all
    # left joins to inner, so we get the expected results
    # read the comment on top of the actual function to see what this does
    $attrs->{from} = $rsrc->schema->storage->_inner_join_to_node ($attrs->{from}, $alias);


    #XXX - temp fix for result_class bug. There likely is a more elegant fix -groditi
    delete @{$attrs}{qw(result_class alias)};

    my $new_cache;

    if (my $cache = $self->get_cache) {
      if ($cache->[0] && $cache->[0]->related_resultset($rel)->get_cache) {
        $new_cache = [ map { @{$_->related_resultset($rel)->get_cache} }
                        @$cache ];
      }
    }

    my $rel_source = $rsrc->related_source($rel);

    my $new = do {

      # The reason we do this now instead of passing the alias to the
      # search_rs below is that if you wrap/overload resultset on the
      # source you need to know what alias it's -going- to have for things
      # to work sanely (e.g. RestrictWithObject wants to be able to add
      # extra query restrictions, and these may need to be $alias.)

      my $rel_attrs = $rel_source->resultset_attributes;
      local $rel_attrs->{alias} = $alias;

      $rel_source->resultset
                 ->search_rs(
                     undef, {
                       %$attrs,
                       where => $attrs->{where},
                   });
    };
    $new->set_cache($new_cache) if $new_cache;
    $new;
  };
}