X-Git-Url: http://git.shadowcat.co.uk/gitweb/gitweb.cgi?p=dbsrgits%2FDBIx-Class.git;a=blobdiff_plain;f=lib%2FDBIx%2FClass%2FResultSet.pm;h=d8dcfcaa45255a3127422b2767426de101daea88;hp=39207b6ab4f099f629b1ef113fe798ed33e60c40;hb=fe0708a2d68b5d34b6bc6f7e70164c3e569f1dd0;hpb=01272eb81fe3a43e0a2f7befa465cc669945d543

diff --git a/lib/DBIx/Class/ResultSet.pm b/lib/DBIx/Class/ResultSet.pm
index 39207b6..d8dcfca 100644
--- a/lib/DBIx/Class/ResultSet.pm
+++ b/lib/DBIx/Class/ResultSet.pm
@@ -2,25 +2,31 @@ package DBIx::Class::ResultSet;
 
 use strict;
 use warnings;
-use overload
-        '0+'     => "count",
-        'bool'   => "_bool",
-        fallback => 1;
-use Carp::Clan qw/^DBIx::Class/;
+use base qw/DBIx::Class/;
+use DBIx::Class::Carp;
 use DBIx::Class::Exception;
-use Data::Page;
-use Storable;
 use DBIx::Class::ResultSetColumn;
-use DBIx::Class::ResultSourceHandle;
+use Scalar::Util qw/blessed weaken/;
+use Try::Tiny;
+use Data::Compare (); # no imports!!! guard against insane architecture
+
+# not importing first() as it will clash with our own method
 use List::Util ();
-use Scalar::Util ();
 
-use base qw/DBIx::Class/;
+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 Test::Deep::NoTest (qw/eq_deeply/);
-use Data::Dumper::Concise;
+use namespace::clean;
 
-__PACKAGE__->mk_group_accessors('simple' => qw/_result_class _source_handle/);
+use overload
+        '0+'     => "count",
+        'bool'   => "_bool",
+        fallback => 1;
+
+__PACKAGE__->mk_group_accessors('simple' => qw/_result_class result_source/);
 
 =head1 NAME
 
@@ -29,6 +35,10 @@ DBIx::Class::ResultSet - Represents a query used for fetching a set of results.
 =head1 SYNOPSIS
 
   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();
 
@@ -57,7 +67,40 @@ represents.
 
 The query that the ResultSet represents is B<only> executed against
 the database when these methods are called:
-L</find> L</next> L</all> L</first> L</single> L</count>
+L</find>, L</next>, L</all>, L</first>, L</single>, L</count>.
+
+If a resultset is used in a numeric context it returns the L</count>.
+However, if it is used in a boolean context it is B<always> true.  So if
+you want to check if a resultset has any results, you must use C<if $rs
+!= 0>.
+
+=head1 CUSTOM ResultSet CLASSES THAT USE Moose
+
+If you want to make your custom ResultSet classes with L<Moose>, use a template
+similar to:
+
+    package MyApp::Schema::ResultSet::User;
+
+    use Moose;
+    use namespace::autoclean;
+    use MooseX::NonMoose;
+    extends 'DBIx::Class::ResultSet';
+
+    sub BUILDARGS { $_[2] }
+
+    ...your code...
+
+    __PACKAGE__->meta->make_immutable;
+
+    1;
+
+The L<MooseX::NonMoose> is necessary so that the L<Moose> constructor does not
+clash with the regular ResultSet constructor. Alternatively, you can use:
+
+    __PACKAGE__->meta->make_immutable(inline_constructor => 0);
+
+The L<BUILDARGS|Moose::Manual::Construction/BUILDARGS> is necessary because the
+signature of the ResultSet C<new> is C<< ->new($source, \%args) >>.
 
 =head1 EXAMPLES
 
@@ -72,14 +115,14 @@ another.
   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 $schema = $self->result_source->schema;
 
     my $cd_rs = $schema->resultset('CD')->search({
       title => $request->param('title'),
       year => $request->param('year'),
     });
 
-    $self->apply_security_policy( $cd_rs );
+    $cd_rs = $self->apply_security_policy( $cd_rs );
 
     return $cd_rs->all();
   }
@@ -101,7 +144,7 @@ attributes with the same keys need resolving.
 L</join>, L</prefetch>, L</+select>, L</+as> attributes are merged
 into the existing ones from the original resultset.
 
-The L</where>, L</having> attribute, and any search conditions are
+The L</where> and L</having> attributes, and any search conditions, are
 merged with an SQL C<AND> to the existing condition from the original
 resultset.
 
@@ -142,13 +185,6 @@ Which is the same as:
 
 See: L</search>, L</count>, L</get_column>, L</all>, L</create>.
 
-=head1 OVERLOADING
-
-If a resultset is used in a numeric context it returns the L</count>.
-However, if it is used in a boolean context it is always true.  So if
-you want to check if a resultset has any results use C<if $rs != 0>.
-C<if $rs> will always be true.
-
 =head1 METHODS
 
 =head2 new
@@ -184,8 +220,8 @@ sub new {
   return $class->new_result(@_) if ref $class;
 
   my ($source, $attrs) = @_;
-  $source = $source->handle
-    unless $source->isa('DBIx::Class::ResultSourceHandle');
+  $source = $source->resolve
+    if $source->isa('DBIx::Class::ResultSourceHandle');
   $attrs = { %{$attrs||{}} };
 
   if ($attrs->{page}) {
@@ -194,23 +230,24 @@ sub new {
 
   $attrs->{alias} ||= 'me';
 
-  # Creation of {} and bless separated to mitigate RH perl bug
-  # see https://bugzilla.redhat.com/show_bug.cgi?id=196836
-  my $self = {
-    _source_handle => $source,
+  my $self = bless {
+    result_source => $source,
     cond => $attrs->{where},
-    count => undef,
     pager => undef,
-    attrs => $attrs
-  };
+    attrs => $attrs,
+  }, $class;
 
-  bless $self, $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->resolve->result_class
+    $attrs->{result_class} || $source->result_class
   );
 
-  return $self;
+  $self;
 }
 
 =head2 search
@@ -219,7 +256,7 @@ sub new {
 
 =item Arguments: $cond, \%attrs?
 
-=item Return Value: $resultset (scalar context), @row_objs (list context)
+=item Return Value: $resultset (scalar context) ||  @row_objs (list context)
 
 =back
 
@@ -229,6 +266,9 @@ sub new {
   my $new_rs = $cd_rs->search([ { year => 2005 }, { year => 2004 } ]);
                  # year = 2005 OR year = 2004
 
+In list context, C<< ->all() >> is called implicitly on the resultset, thus
+returning a list of row objects instead. To avoid that, use L</search_rs>.
+
 If you need to pass in additional attributes but no additional condition,
 call it as C<search(undef, \%attrs)>.
 
@@ -240,16 +280,44 @@ call it as C<search(undef, \%attrs)>.
 For a list of attributes that can be passed to C<search>, see
 L</ATTRIBUTES>. For more examples of using this function, see
 L<Searching|DBIx::Class::Manual::Cookbook/Searching>. For a complete
-documentation for the first argument, see L<SQL::Abstract>.
+documentation for the first argument, see L<SQL::Abstract>
+and its extension L<DBIx::Class::SQLMaker>.
 
 For more help on using joins with search, see L<DBIx::Class::Manual::Joining>.
 
+=head3 CAVEAT
+
+Note that L</search> does not process/deflate any of the values passed in the
+L<SQL::Abstract>-compatible search condition structure. This is unlike other
+condition-bound methods L</new>, L</create> and L</find>. The user must ensure
+manually that any value passed to this method will stringify to something the
+RDBMS knows how to deal with. A notable example is the handling of L<DateTime>
+objects, for more info see:
+L<DBIx::Class::Manual::Cookbook/Formatting DateTime objects in queries>.
+
 =cut
 
 sub search {
   my $self = shift;
   my $rs = $self->search_rs( @_ );
-  return (wantarray ? $rs->all : $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 ();
+  }
 }
 
 =head2 search_rs
@@ -272,113 +340,288 @@ sub search_rs {
 
   # Special-case handling for (undef, undef).
   if ( @_ == 2 && !defined $_[1] && !defined $_[0] ) {
-    pop(@_); pop(@_);
+    @_ = ();
   }
 
-  my $attrs = {};
-  $attrs = pop(@_) if @_ > 1 and ref $_[$#_] eq 'HASH';
-  my $our_attrs = { %{$self->{attrs}} };
-  my $having = delete $our_attrs->{having};
-  my $where = delete $our_attrs->{where};
-
-  my $rows;
+  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;
+  }
 
-  unless (
-    (@_ && defined($_[0])) # @_ == () or (undef)
-    ||
-    (keys %$attrs # empty attrs or only 'safe' attrs
-    && List::Util::first { !$safe{$_} } keys %$attrs)
-  ) {
-    # no search, effectively just a clone
-    $rows = $self->get_cache;
-  }
-
-  # reset the selector list
-  if (List::Util::first { exists $attrs->{$_} } qw{columns select as}) {
-     delete @{$our_attrs}{qw{select as columns +select +as +columns include_columns}};
-  }
-
-  my $new_attrs = { %{$our_attrs}, %{$attrs} };
-
-  # merge new attrs into inherited
-  foreach my $key (qw/join prefetch +select +as +columns include_columns bind/) {
-    next unless exists $attrs->{$key};
-    $new_attrs->{$key} = $self->_merge_attr($our_attrs->{$key}, $attrs->{$key});
-  }
-
-  my $cond = (@_
-    ? (
-        (@_ == 1 || ref $_[0] eq "HASH")
-          ? (
-              (ref $_[0] eq 'HASH')
-                ? (
-                    (keys %{ $_[0] }  > 0)
-                      ? shift
-                      : undef
-                   )
-                :  shift
-             )
-          : (
-              (@_ % 2)
-                ? $self->throw_exception("Odd number of arguments to search")
-                : {@_}
-             )
-      )
-    : undef
-  );
+  my $rsrc = $self->result_source;
 
-  if (defined $where) {
-    $new_attrs->{where} = (
-      defined $new_attrs->{where}
-        ? { '-and' => [
-              map {
-                ref $_ eq 'ARRAY' ? [ -or => $_ ] : $_
-              } $where, $new_attrs->{where}
-            ]
-          }
-        : $where);
+  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} || [] } ];
   }
 
-  if (defined $cond) {
-    $new_attrs->{where} = (
-      defined $new_attrs->{where}
-        ? { '-and' => [
-              map {
-                ref $_ eq 'ARRAY' ? [ -or => $_ ] : $_
-              } $cond, $new_attrs->{where}
-            ]
-          }
-        : $cond);
+
+  # 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';
   }
 
-  if (defined $having) {
-    $new_attrs->{having} = (
-      defined $new_attrs->{having}
-        ? { '-and' => [
-              map {
-                ref $_ eq 'ARRAY' ? [ -or => $_ ] : $_
-              } $having, $new_attrs->{having}
-            ]
-          }
-        : $having);
+  for ($old_where, $call_cond) {
+    if (defined $_) {
+      $new_attrs->{where} = $self->_stack_cond (
+        $_, $new_attrs->{where}
+      );
+    }
   }
 
-  my $rs = (ref $self)->new($self->result_source, $new_attrs);
+  if (defined $old_having) {
+    $new_attrs->{having} = $self->_stack_cond (
+      $old_having, $new_attrs->{having}
+    )
+  }
 
-  $rs->set_cache($rows) if ($rows);
+  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) = @_;
+
+  # collapse single element top-level conditions
+  # (single pass only, unlikely to need recursion)
+  for ($left, $right) {
+    if (ref $_ eq 'ARRAY') {
+      if (@$_ == 0) {
+        $_ = undef;
+      }
+      elsif (@$_ == 1) {
+        $_ = $_->[0];
+      }
+    }
+    elsif (ref $_ eq 'HASH') {
+      my ($first, $more) = keys %$_;
+
+      # empty hash
+      if (! defined $first) {
+        $_ = undef;
+      }
+      # one element hash
+      elsif (! defined $more) {
+        if ($first eq '-and' and ref $_->{'-and'} eq 'HASH') {
+          $_ = $_->{'-and'};
+        }
+        elsif ($first eq '-or' and ref $_->{'-or'} eq 'ARRAY') {
+          $_ = $_->{'-or'};
+        }
+      }
+    }
+  }
+
+  # merge hashes with weeding out of duplicates (simple cases only)
+  if (ref $left eq 'HASH' and ref $right eq 'HASH') {
+
+    # shallow copy to destroy
+    $right = { %$right };
+    for (grep { exists $right->{$_} } keys %$left) {
+      # the use of eq_deeply here is justified - the rhs of an
+      # expression can contain a lot of twisted weird stuff
+      delete $right->{$_} if Data::Compare::Compare( $left->{$_}, $right->{$_} );
+    }
+
+    $right = undef unless keys %$right;
+  }
+
+
+  if (defined $left xor defined $right) {
+    return defined $left ? $left : $right;
+  }
+  elsif (! defined $left) {
+    return undef;
+  }
+  else {
+    return { -and => [ $left, $right ] };
+  }
+}
+
 =head2 search_literal
 
 =over 4
 
 =item Arguments: $sql_fragment, @bind_values
 
-=item Return Value: $resultset (scalar context), @row_objs (list context)
+=item Return Value: $resultset (scalar context) || @row_objs (list context)
 
 =back
 
@@ -418,25 +661,56 @@ sub search_literal {
 
 =over 4
 
-=item Arguments: @values | \%cols, \%attrs?
+=item Arguments: \%columns_values | @pk_values, \%attrs?
 
 =item Return Value: $row_object | undef
 
 =back
 
-Finds a row based on its primary key or unique constraint. For example, to find
-a row by its primary key:
+Finds and returns a single row based on supplied criteria. Takes either a
+hashref with the same format as L</create> (including inference of foreign
+keys from related objects), or a list of primary key values in the same
+order as the L<primary columns|DBIx::Class::ResultSource/primary_columns>
+declaration on the L</result_source>.
+
+In either case an attempt is made to combine conditions already existing on
+the resultset with the condition passed to this method.
+
+To aid with preparing the correct query for the storage you may supply the
+C<key> attribute, which is the name of a
+L<unique constraint|DBIx::Class::ResultSource/add_unique_constraint> (the
+unique constraint corresponding to the
+L<primary columns|DBIx::Class::ResultSource/primary_columns> is always named
+C<primary>). If the C<key> attribute has been supplied, and DBIC is unable
+to construct a query that satisfies the named unique constraint fully (
+non-NULL values for each column member of the constraint) an exception is
+thrown.
+
+If no C<key> is specified, the search is carried over all unique constraints
+which are fully defined by the available condition.
+
+If no such constraint is found, C<find> currently defaults to a simple
+C<< search->(\%column_values) >> which may or may not do what you expect.
+Note that this fallback behavior may be deprecated in further versions. If
+you need to search with arbitrary conditions - use L</search>. If the query
+resulting from this fallback produces more than one row, a warning to the
+effect is issued, though only the first row is constructed and returned as
+C<$row_object>.
 
-  my $cd = $schema->resultset('CD')->find(5);
+In addition to C<key>, L</find> recognizes and applies standard
+L<resultset attributes|/ATTRIBUTES> in the same way as L</search> does.
 
-You can also find a row by a specific unique constraint using the C<key>
-attribute. For example:
+Note that if you have extra concerns about the correctness of the resulting
+query you need to specify the C<key> attribute and supply the entire condition
+as an argument to find (since it is not always possible to perform the
+combination of the resultset condition with the supplied one, especially if
+the resultset condition contains literal sql).
 
-  my $cd = $schema->resultset('CD')->find('Massive Attack', 'Mezzanine', {
-    key => 'cd_artist_title'
-  });
+For example, to find a row by its primary key:
+
+  my $cd = $schema->resultset('CD')->find(5);
 
-Additionally, you can specify the columns explicitly by name:
+You can also find a row by a specific unique constraint:
 
   my $cd = $schema->resultset('CD')->find(
     {
@@ -446,24 +720,7 @@ Additionally, you can specify the columns explicitly by name:
     { key => 'cd_artist_title' }
   );
 
-If the C<key> is specified as C<primary>, it searches only on the primary key.
-
-If no C<key> is specified, it searches on all unique constraints defined on the
-source for which column data is provided, including the primary key.
-
-If your table does not have a primary key, you B<must> provide a value for the
-C<key> attribute matching one of the unique constraints on the source.
-
-In addition to C<key>, L</find> recognizes and applies standard
-L<resultset attributes|/ATTRIBUTES> in the same way as L</search> does.
-
-Note: If your query does not return only one row, a warning is generated:
-
-  Query returned more than one row
-
-See also L</find_or_create> and L</update_or_create>. For information on how to
-declare unique constraints, see
-L<DBIx::Class::ResultSource/add_unique_constraint>.
+See also L</find_or_create> and L</update_or_create>.
 
 =cut
 
@@ -471,57 +728,75 @@ sub find {
   my $self = shift;
   my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
 
-  # Default to the primary key, but allow a specific key
-  my @cols = exists $attrs->{key}
-    ? $self->result_source->unique_constraint_columns($attrs->{key})
-    : $self->result_source->primary_columns;
-  $self->throw_exception(
-    "Can't find unless a primary key is defined or unique constraint is specified"
-  ) unless @cols;
+  my $rsrc = $self->result_source;
+
+  my $constraint_name;
+  if (exists $attrs->{key}) {
+    $constraint_name = defined $attrs->{key}
+      ? $attrs->{key}
+      : $self->throw_exception("An undefined 'key' resultset attribute makes no sense")
+    ;
+  }
+
+  # Parse out the condition from input
+  my $call_cond;
 
-  # Parse out a hashref from input
-  my $input_query;
   if (ref $_[0] eq 'HASH') {
-    $input_query = { %{$_[0]} };
-  }
-  elsif (@_ == @cols) {
-    $input_query = {};
-    @{$input_query}{@cols} = @_;
+    $call_cond = { %{$_[0]} };
   }
   else {
-    # Compatibility: Allow e.g. find(id => $value)
-    carp "Find by key => value deprecated; please use a hashref instead";
-    $input_query = {@_};
-  }
-
-  my (%related, $info);
-
-  KEY: foreach my $key (keys %$input_query) {
-    if (ref($input_query->{$key})
-        && ($info = $self->result_source->relationship_info($key))) {
-      my $val = delete $input_query->{$key};
-      next KEY if (ref($val) eq 'ARRAY'); # has_many for multi_create
-      my $rel_q = $self->result_source->_resolve_condition(
-                    $info->{cond}, $val, $key
-                  );
-      die "Can't handle OR join condition in find" if ref($rel_q) eq 'ARRAY';
+    # if only values are supplied we need to default to 'primary'
+    $constraint_name = 'primary' unless defined $constraint_name;
+
+    my @c_cols = $rsrc->unique_constraint_columns($constraint_name);
+
+    $self->throw_exception(
+      "No constraint columns, maybe a malformed '$constraint_name' 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_name'"
+    ) 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;
     }
   }
-  if (my @keys = keys %related) {
-    @{$input_query}{@keys} = values %related;
-  }
 
+  # relationship conditions take precedence (?)
+  @{$call_cond}{keys %related} = values %related;
 
-  # Build the final query: Default to the disjunction of the unique queries,
-  # but allow the input query in case the ResultSet defines the query or the
-  # user is abusing find
   my $alias = exists $attrs->{alias} ? $attrs->{alias} : $self->{attrs}{alias};
-  my $query;
-  if (exists $attrs->{key}) {
-    my @unique_cols = $self->result_source->unique_constraint_columns($attrs->{key});
-    my $unique_query = $self->_build_unique_query($input_query, \@unique_cols);
-    $query = $self->_add_alias($unique_query, $alias);
+  my $final_cond;
+  if (defined $constraint_name) {
+    $final_cond = $self->_qualify_cond_columns (
+
+      $self->_build_unique_cond (
+        $constraint_name,
+        $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
@@ -532,14 +807,28 @@ sub find {
     # relationship
   }
   else {
-    my @unique_queries = $self->_unique_queries($input_query, $attrs);
-    $query = @unique_queries
-      ? [ map { $self->_add_alias($_, $alias) } @unique_queries ]
-      : $self->_add_alias($input_query, $alias);
+    # 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
-  my $rs = $self->search ($query, {result_class => $self->result_class, %$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 ($rs->_resolved_attrs->{collapse}) {
     my $row = $rs->next;
     carp "Query returned more than one row" if $rs->next;
@@ -550,80 +839,97 @@ sub find {
   }
 }
 
-# _add_alias
+# 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
 #
-# Add the specified alias to the specified query hash. A copy is made so the
-# original query is not modified.
+# *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 _add_alias {
-  my ($self, $query, $alias) = @_;
 
-  my %aliased = %$query;
-  foreach my $col (grep { ! m/\./ } keys %aliased) {
-    $aliased{"$alias.$col"} = delete $aliased{$col};
+sub _qualify_cond_columns {
+  my ($self, $cond, $alias) = @_;
+
+  my %aliased = %$cond;
+  for (keys %aliased) {
+    $aliased{"$alias.$_"} = delete $aliased{$_}
+      if $_ !~ /\./;
   }
 
   return \%aliased;
 }
 
-# _unique_queries
-#
-# Build a list of queries which satisfy unique constraints.
-
-sub _unique_queries {
-  my ($self, $query, $attrs) = @_;
-
-  my @constraint_names = exists $attrs->{key}
-    ? ($attrs->{key})
-    : $self->result_source->unique_constraint_names;
+sub _build_unique_cond {
+  my ($self, $constraint_name, $extra_cond, $croak_on_null) = @_;
 
-  my $where = $self->_collapse_cond($self->{attrs}{where} || {});
-  my $num_where = scalar keys %$where;
+  my @c_cols = $self->result_source->unique_constraint_columns($constraint_name);
 
-  my (@unique_queries, %seen_column_combinations);
-  foreach my $name (@constraint_names) {
-    my @constraint_cols = $self->result_source->unique_constraint_columns($name);
-
-    my $constraint_sig = join "\x00", sort @constraint_cols;
-    next if $seen_column_combinations{$constraint_sig}++;
-
-    my $unique_query = $self->_build_unique_query($query, \@constraint_cols);
+  # combination may fail if $self->{cond} is non-trivial
+  my ($final_cond) = try {
+    $self->_merge_with_rscond ($extra_cond)
+  } catch {
+    +{ %$extra_cond }
+  };
 
-    my $num_cols = scalar @constraint_cols;
-    my $num_query = scalar keys %$unique_query;
+  # trim out everything not in $columns
+  $final_cond = { map {
+    exists $final_cond->{$_}
+      ? ( $_ => $final_cond->{$_} )
+      : ()
+  } @c_cols };
 
-    my $total = $num_query + $num_where;
-    if ($num_query && ($num_query == $num_cols || $total == $num_cols)) {
-      # The query is either unique on its own or is unique in combination with
-      # the existing where clause
-      push @unique_queries, $unique_query;
-    }
+  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),
+    ) );
   }
 
-  return @unique_queries;
-}
-
-# _build_unique_query
-#
-# Constrain the specified query hash based on the specified column names.
-
-sub _build_unique_query {
-  my ($self, $query, $unique_cols) = @_;
+  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 {
-    map  { $_ => $query->{$_} }
-    grep { exists $query->{$_} }
-      @$unique_cols
-  };
+  return $final_cond;
 }
 
 =head2 search_related
 
 =over 4
 
-=item Arguments: $rel, $cond, \%attrs?
+=item Arguments: $rel, $cond?, \%attrs?
 
-=item Return Value: $new_resultset
+=item Return Value: $new_resultset (scalar context) || @row_objs (list context)
 
 =back
 
@@ -634,6 +940,11 @@ sub _build_unique_query {
 Searches the specified relationship, optionally specifying a condition and
 attributes for matching records. See L</ATTRIBUTES> for more information.
 
+In list context, C<< ->all() >> is called implicitly on the resultset, thus
+returning a list of row objects instead. To avoid that, use L</search_related_rs>.
+
+See also L</search_related_rs>.
+
 =cut
 
 sub search_related {
@@ -682,15 +993,15 @@ sub cursor {
 
 =item Arguments: $cond?
 
-=item Return Value: $row_object?
+=item Return Value: $row_object | undef
 
 =back
 
   my $cd = $schema->resultset('CD')->single({ year => 2001 });
 
 Inflates the first result without creating a cursor if the resultset has
-any records in it; if not returns nothing. Used by L</find> as a lean version of
-L</search>.
+any records in it; if not returns C<undef>. Used by L</find> as a lean version
+of L</search>.
 
 While this method can take an optional search condition (just like L</search>)
 being a fast-code-path it does not recognize search attributes. If you need to
@@ -745,12 +1056,6 @@ sub single {
     }
   }
 
-#  XXX: Disabled since it doesn't infer uniqueness in all cases
-#  unless ($self->_is_unique_query($attrs->{where})) {
-#    carp "Query not guaranteed to return a single row"
-#      . "; please declare your unique constraints or use search instead";
-#  }
-
   my @data = $self->result_source->storage->select_single(
     $attrs->{from}, $attrs->{select},
     $attrs->{where}, $attrs
@@ -763,38 +1068,6 @@ sub single {
 }
 
 
-# _is_unique_query
-#
-# Try to determine if the specified query is guaranteed to be unique, based on
-# the declared unique constraints.
-
-sub _is_unique_query {
-  my ($self, $query) = @_;
-
-  my $collapsed = $self->_collapse_query($query);
-  my $alias = $self->{attrs}{alias};
-
-  foreach my $name ($self->result_source->unique_constraint_names) {
-    my @unique_cols = map {
-      "$alias.$_"
-    } $self->result_source->unique_constraint_columns($name);
-
-    # Count the values for each unique column
-    my %seen = map { $_ => 0 } @unique_cols;
-
-    foreach my $key (keys %$collapsed) {
-      my $aliased = $key =~ /\./ ? $key : "$alias.$key";
-      next unless exists $seen{$aliased};  # Additional constraints are okay
-      $seen{$aliased} = scalar keys %{ $collapsed->{$key} };
-    }
-
-    # If we get 0 or more than 1 value for a column, it's not necessarily unique
-    return 1 unless grep { $_ != 1 } values %seen;
-  }
-
-  return 0;
-}
-
 # _collapse_query
 #
 # Recursively collapse the query, accumulating values for each column.
@@ -855,7 +1128,7 @@ sub get_column {
 
 =item Arguments: $cond, \%attrs?
 
-=item Return Value: $resultset (scalar context), @row_objs (list context)
+=item Return Value: $resultset (scalar context) || @row_objs (list context)
 
 =back
 
@@ -881,7 +1154,7 @@ instead. An example conversion is:
 
 sub search_like {
   my $class = shift;
-  carp (
+  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!)'
@@ -898,7 +1171,7 @@ sub search_like {
 
 =item Arguments: $first, $last
 
-=item Return Value: $resultset (scalar context), @row_objs (list context)
+=item Return Value: $resultset (scalar context) || @row_objs (list context)
 
 =back
 
@@ -916,7 +1189,7 @@ sub slice {
   $attrs->{offset} = $self->{attrs}{offset} || 0;
   $attrs->{offset} += $min;
   $attrs->{rows} = ($max ? ($max - $min + 1) : 1);
-  return $self->search(undef(), $attrs);
+  return $self->search(undef, $attrs);
   #my $slice = (ref $self)->new($self->result_source, $attrs);
   #return (wantarray ? $slice->all : $slice);
 }
@@ -927,7 +1200,7 @@ sub slice {
 
 =item Arguments: none
 
-=item Return Value: $result?
+=item Return Value: $result | undef
 
 =back
 
@@ -953,6 +1226,7 @@ sub next {
     return $cache->[$self->{all_cache_position}++];
   }
   if ($self->{attrs}{cache}) {
+    delete $self->{pager};
     $self->{all_cache_position} = 1;
     return ($self->all)[0];
   }
@@ -1100,7 +1374,6 @@ sub _check_register {
   return undef;
 }
 
-
 sub _merge_result {
   my ( $self, $result, $row, $register ) = @_;
   return @$result = @$row if ( @$result == 0 );  # initialize with $row
@@ -1108,8 +1381,7 @@ sub _merge_result {
   my ( undef, $rels,   $ids )   = @$result;
   my ( undef, $new_rels, $new_ids ) = @$row;
 
-  use List::MoreUtils;
-  my @rels = List::MoreUtils::uniq( keys %$rels, keys %$new_rels );
+  my @rels = keys %{ { %{$rels||{} }, %{ $new_rels||{} } } };
   foreach my $rel (@rels) {
     $register = $register->{$rel} ||= {};
 
@@ -1130,9 +1402,6 @@ sub _merge_result {
   return 1;
 }
 
-
-
-
 =begin
 
 # two arguments: $as_proto is an arrayref of column names,
@@ -1366,9 +1635,14 @@ in the original source class will not run.
 sub result_class {
   my ($self, $result_class) = @_;
   if ($result_class) {
-    $self->ensure_class_loaded($result_class);
+    unless (ref $result_class) { # don't fire this for an object
+      $self->ensure_class_loaded($result_class);
+    }
     $self->_result_class($result_class);
-    $self->{attrs}{result_class} = $result_class if ref $self;
+    # 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;
 }
@@ -1464,13 +1738,13 @@ sub _count_rs {
   $attrs ||= $self->_resolved_attrs;
 
   my $tmp_attrs = { %$attrs };
-
-  # take off any limits, record_filter is cdbi, and no point of ordering a count
-  delete $tmp_attrs->{$_} for (qw/select as rows offset order_by record_filter/);
+  # 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, $tmp_attrs);
+  $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');
 
@@ -1484,37 +1758,91 @@ sub _count_subq_rs {
   my ($self, $attrs) = @_;
 
   my $rsrc = $self->result_source;
-  $attrs ||= $self->_resolved_attrs_copy;
+  $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/};
 
-  # extra selectors do not go in the subquery and there is no point of ordering it
-  delete $sub_attrs->{$_} for qw/collapse select _prefetch_select as order_by/;
-
-  # if we multi-prefetch we group_by primary keys only as this is what we would
+  # if we multi-prefetch we group_by something unique, as this is what we would
   # get out of the rs via ->next/->all. We *DO WANT* to clobber old group_by regardless
   if ( $attrs->{collapse}  ) {
-    $sub_attrs->{group_by} = [ map { "$attrs->{alias}.$_" } ($rsrc->_pri_cols) ]
+    $sub_attrs->{group_by} = [ map { "$attrs->{alias}.$_" } @{
+      $rsrc->_identifying_column_set || $self->throw_exception(
+        'Unable to construct a unique group_by criteria properly collapsing the '
+      . 'has_many prefetch before count()'
+      );
+    } ]
   }
 
-  $sub_attrs->{select} = $rsrc->storage->_subq_count_select ($rsrc, $attrs);
+  # Calculate subquery selector
+  if (my $g = $sub_attrs->{group_by}) {
 
-  # this is so that the query can be simplified e.g.
-  # * ordering can be thrown away in things like Top limit
-  $sub_attrs->{-for_count_only} = 1;
+    my $sql_maker = $rsrc->storage->sql_maker;
 
-  my $sub_rs = $rsrc->resultset_class->new ($rsrc, $sub_attrs);
+    # 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});
+    }
 
-  $attrs->{from} = [{
-    -alias => 'count_subq',
-    -source_handle => $rsrc->handle,
-    count_subq => $sub_rs->as_query,
-  }];
+    # 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} = '';
+      }
 
-  # the subquery replaces this
-  delete $attrs->{$_} for qw/where bind collapse group_by having having_bind rows offset/;
+      my ($lquote, $rquote, $sep) = map { quotemeta $_ } ($sql_maker->_quote_chars, $sql_maker->name_sep);
 
-  return $self->_count_rs ($attrs);
+      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 {
@@ -1548,8 +1876,7 @@ sub count_literal { shift->search_literal(@_)->count; }
 
 =back
 
-Returns all elements in the resultset. Called implicitly if the resultset
-is returned in list context.
+Returns all elements in the resultset.
 
 =cut
 
@@ -1617,12 +1944,12 @@ sub reset {
 
 =item Arguments: none
 
-=item Return Value: $object?
+=item Return Value: $object | undef
 
 =back
 
-Resets the resultset and returns an object for the first result (if the
-resultset returns anything).
+Resets the resultset and returns an object for the first result (or C<undef>
+if the resultset is empty).
 
 =cut
 
@@ -1640,38 +1967,122 @@ sub first {
 sub _rs_update_delete {
   my ($self, $op, $values) = @_;
 
+  my $cond = $self->{cond};
   my $rsrc = $self->result_source;
+  my $storage = $rsrc->schema->storage;
 
-  # if a condition exists we need to strip all table qualifiers
-  # if this is not possible we'll force a subquery below
-  my $cond = $rsrc->schema->storage->_strip_cond_qualifiers ($self->{cond});
+  my $attrs = { %{$self->_resolved_attrs} };
 
-  my $needs_group_by_subq = $self->_has_resolved_attr (qw/collapse group_by -join/);
-  my $needs_subq = $needs_group_by_subq || (not defined $cond) || $self->_has_resolved_attr(qw/row offset/);
+  # "needs" is a strong word here - if the subquery is part of an IN clause - no point of
+  # even adding the group_by. It will really be used only when composing a poor-man's
+  # multicolumn-IN equivalent OR set
+  my $needs_group_by_subq = defined $attrs->{group_by};
 
-  if ($needs_group_by_subq or $needs_subq) {
+  # simplify the joinmap and maybe decide if a grouping (and thus subquery) is necessary
+  my $relation_classifications;
+  if (ref($attrs->{from}) eq 'ARRAY') {
+    $attrs->{from} = $storage->_prune_unused_joins ($attrs->{from}, $attrs->{select}, $cond, $attrs);
 
-    # make a new $rs selecting only the PKs (that's all we really need)
-    my $attrs = $self->_resolved_attrs_copy;
+    $relation_classifications = $storage->_resolve_aliastypes_from_select_args (
+      [ @{$attrs->{from}}[1 .. $#{$attrs->{from}}] ],
+      $attrs->{select},
+      $cond,
+      $attrs
+    ) unless $needs_group_by_subq;  # we already know we need a group, no point of resolving them
+  }
+  else {
+    $needs_group_by_subq ||= 1; # if {from} is unparseable assume the worst
+  }
 
-    delete $attrs->{$_} for qw/collapse select as/;
-    $attrs->{columns} = [ map { "$attrs->{alias}.$_" } ($self->result_source->_pri_cols) ];
+  $needs_group_by_subq ||= exists $relation_classifications->{multiplying};
 
+  # if no subquery - life is easy-ish
+  unless (
+    $needs_group_by_subq
+      or
+    keys %$relation_classifications # if any joins at all - need to wrap a subq
+      or
+    $self->_has_resolved_attr(qw/rows offset/) # limits call for a subq
+  ) {
+    # 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] : (),
+    );
+  }
+
+  # we got this far - means it is time to wrap a subquery
+  my $idcols = $rsrc->_identifying_column_set || $self->throw_exception(
+    sprintf(
+      "Unable to perform complex resultset %s() without an identifying set of columns on source '%s'",
+      $op,
+      $rsrc->source_name,
+    )
+  );
+  my $existing_group_by = delete $attrs->{group_by};
+
+  # make a new $rs selecting only the PKs (that's all we really need for the subq)
+  delete $attrs->{$_} for qw/collapse _collapse_order_by select _prefetch_selector_range as/;
+  $attrs->{columns} = [ map { "$attrs->{alias}.$_" } @$idcols ];
+  $attrs->{group_by} = \ '';  # FIXME - this is an evil hack, it causes the optimiser to kick in and throw away the LEFT joins
+  my $subrs = (ref $self)->new($rsrc, $attrs);
+
+  if (@$idcols == 1) {
+    return $storage->$op (
+      $rsrc,
+      $op eq 'update' ? $values : (),
+      { $idcols->[0] => { -in => $subrs->as_query } },
+    );
+  }
+  elsif ($storage->_use_multicolumn_in) {
+    # This is hideously ugly, but SQLA does not understand multicol IN expressions
+    my $sql_maker = $storage->sql_maker;
+    my ($sql, @bind) = @${$subrs->as_query};
+    $sql = sprintf ('(%s) IN %s', # the as_query already comes with a set of parenthesis
+      join (', ', map { $sql_maker->_quote ($_) } @$idcols),
+      $sql,
+    );
+
+    return $storage->$op (
+      $rsrc,
+      $op eq 'update' ? $values : (),
+      \[$sql, @bind],
+    );
+  }
+  else {
+    # if all else fails - get all primary keys and operate over a ORed set
+    # wrap in a transaction for consistency
+    # this is where the group_by starts to matter
+    my $subq_group_by;
     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
+      $subq_group_by = $attrs->{columns};
 
-      if (my $g = $attrs->{group_by}) {
+      # make sure if there is a supplied group_by it matches the columns compiled above
+      # perfectly. Anything else can not be sanely executed on most databases so croak
+      # right then and there
+      if ($existing_group_by) {
         my @current_group_by = map
           { $_ =~ /\./ ? $_ : "$attrs->{alias}.$_" }
-          @$g
+          @$existing_group_by
         ;
 
         if (
           join ("\x00", sort @current_group_by)
             ne
-          join ("\x00", sort @{$attrs->{columns}} )
+          join ("\x00", sort @$subq_group_by )
         ) {
           $self->throw_exception (
             "You have just attempted a $op operation on a resultset which does group_by"
@@ -1682,21 +2093,27 @@ sub _rs_update_delete {
           );
         }
       }
-      else {
-        $attrs->{group_by} = $attrs->{columns};
-      }
     }
 
-    my $subrs = (ref $self)->new($rsrc, $attrs);
+    my $guard = $storage->txn_scope_guard;
 
-    return $self->result_source->storage->_subq_update_delete($subrs, $op, $values);
-  }
-  else {
-    return $rsrc->storage->$op(
+    my @op_condition;
+    for my $row ($subrs->search({}, { group_by => $subq_group_by })->cursor->all) {
+      push @op_condition, { map
+        { $idcols->[$_] => $row->[$_] }
+        (0 .. $#$idcols)
+      };
+    }
+
+    my $res = $storage->$op (
       $rsrc,
       $op eq 'update' ? $values : (),
-      $cond,
+      \@op_condition,
     );
+
+    $guard->commit;
+
+    return $res;
   }
 }
 
@@ -1711,8 +2128,25 @@ sub _rs_update_delete {
 =back
 
 Sets the specified columns in the resultset to the supplied values in a
-single query. Return value will be true if the update succeeded or false
-if no records were updated; exact type of success value is storage-dependent.
+single query. Note that this will not run any accessor/set_column/update
+triggers, nor will it update any row object instances derived from this
+resultset (this includes the contents of the L<resultset cache|/set_cache>
+if any). See L</update_all> if you need to execute any on-update
+triggers or cascades defined either by you or a
+L<result component|DBIx::Class::Manual::Component/WHAT IS A COMPONENT>.
+
+The return value is a pass through of what the underlying
+storage backend returned, and may vary. See L<DBI/execute> for the most
+common case.
+
+=head3 CAVEAT
+
+Note that L</update> does not process/deflate any of the values passed in.
+This is unlike the corresponding L<DBIx::Class::Row/update>. The user must
+ensure manually that any value passed to this method will stringify to
+something the RDBMS knows how to deal with. A notable example is the
+handling of L<DateTime> objects, for more info see:
+L<DBIx::Class::Manual::Cookbook/Formatting DateTime objects in queries>.
 
 =cut
 
@@ -1734,8 +2168,9 @@ sub update {
 
 =back
 
-Fetches all objects and updates them one at a time. Note that C<update_all>
-will run DBIC cascade triggers, while L</update> will not.
+Fetches all objects and updates them one at a time via
+L<DBIx::Class::Row/update>. Note that C<update_all> will run DBIC defined
+triggers, while L</update> will not.
 
 =cut
 
@@ -1743,9 +2178,10 @@ sub update_all {
   my ($self, $values) = @_;
   $self->throw_exception('Values for update_all must be a hash')
     unless ref $values eq 'HASH';
-  foreach my $obj ($self->all) {
-    $obj->set_columns($values)->update;
-  }
+
+  my $guard = $self->result_source->schema->txn_scope_guard;
+  $_->update({%$values}) for $self->all;  # shallow copy - update will mangle it
+  $guard->commit;
   return 1;
 }
 
@@ -1759,12 +2195,16 @@ sub update_all {
 
 =back
 
-Deletes the contents of the resultset from its result source. Note that this
-will not run DBIC cascade triggers. See L</delete_all> if you need triggers
-to run. See also L<DBIx::Class::Row/delete>.
+Deletes the rows matching this resultset in a single query. Note that this
+will not run any delete triggers, nor will it alter the
+L<in_storage|DBIx::Class::Row/in_storage> status of any row object instances
+derived from this resultset (this includes the contents of the
+L<resultset cache|/set_cache> if any). See L</delete_all> if you need to
+execute any on-delete triggers or cascades defined either by you or a
+L<result component|DBIx::Class::Manual::Component/WHAT IS A COMPONENT>.
 
-Return value will be the amount of rows deleted; exact type of return value
-is storage-dependent.
+The return value is a pass through of what the underlying storage backend
+returned, and may vary. See L<DBI/execute> for the most common case.
 
 =cut
 
@@ -1786,8 +2226,9 @@ sub delete {
 
 =back
 
-Fetches all objects and deletes them one at a time. Note that C<delete_all>
-will run DBIC cascade triggers, while L</delete> will not.
+Fetches all objects and deletes them one at a time via
+L<DBIx::Class::Row/delete>. Note that C<delete_all> will run DBIC defined
+triggers, while L</delete> will not.
 
 =cut
 
@@ -1796,7 +2237,9 @@ sub delete_all {
   $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;
 }
 
@@ -1810,7 +2253,7 @@ sub delete_all {
 
 Accepts either an arrayref of hashrefs or alternatively an arrayref of arrayrefs.
 For the arrayref of hashrefs style each hashref should be a structure suitable
-forsubmitting to a $resultset->create(...) method.
+for submitting to a $resultset->create(...) method.
 
 In void context, C<insert_bulk> in L<DBIx::Class::Storage::DBI> is used
 to insert the data, as this is a faster method.
@@ -1879,27 +2322,32 @@ sub populate {
   # cruft placed in standalone method
   my $data = $self->_normalize_populate_args(@_);
 
+  return unless @$data;
+
   if(defined wantarray) {
     my @created;
     foreach my $item (@$data) {
       push(@created, $self->create($item));
     }
     return wantarray ? @created : \@created;
-  } else {
+  }
+  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->{$_};
-      $self->result_source->has_relationship($_) && ($ref eq 'ARRAY' or $ref eq 'HASH')
+      $rels->{$_} && ($ref eq 'ARRAY' or $ref eq 'HASH')
         ? push @rels, $_
         : push @columns, $_
       ;
     }
 
-    my @pks = $self->result_source->primary_columns;
+    my @pks = $rsrc->primary_columns;
 
     ## do the belongs_to relationships
     foreach my $index (0..$#$data) {
@@ -1917,11 +2365,12 @@ sub populate {
       foreach my $rel (@rels) {
         next unless ref $data->[$index]->{$rel} eq "HASH";
         my $result = $self->related_resultset($rel)->create($data->[$index]->{$rel});
-        my ($reverse) = keys %{$self->result_source->reverse_relationship_info($rel)};
+        my ($reverse_relname, $reverse_relinfo) = %{$rsrc->reverse_relationship_info($rel)};
         my $related = $result->result_source->_resolve_condition(
-          $result->result_source->relationship_info($reverse)->{cond},
+          $reverse_relinfo->{cond},
           $self,
           $result,
+          $rel,
         );
 
         delete $data->[$index]->{$rel};
@@ -1932,14 +2381,14 @@ sub populate {
     }
 
     ## inherit the data locked in the conditions of the resultset
-    my ($rs_data) = $self->_merge_cond_with_data({});
+    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
-    $self->result_source->storage->insert_bulk(
-      $self->result_source,
+    $rsrc->storage->insert_bulk(
+      $rsrc,
       [@columns, @inherit_cols],
       [ map { [ @$_{@columns}, @inherit_data ] } @$data ],
     );
@@ -1947,18 +2396,20 @@ sub populate {
     ## do the has_many relationships
     foreach my $item (@$data) {
 
+      my $main_row;
+
       foreach my $rel (@rels) {
-        next unless $item->{$rel} && ref $item->{$rel} eq "ARRAY";
+        next unless ref $item->{$rel} eq "ARRAY" && @{ $item->{$rel} };
 
-        my $parent = $self->find({map { $_ => $item->{$_} } @pks})
-     || $self->throw_exception('Cannot find the relating object.');
+        $main_row ||= $self->new_result({map { $_ => $item->{$_} } @pks});
 
-        my $child = $parent->$rel;
+        my $child = $main_row->$rel;
 
         my $related = $child->result_source->_resolve_condition(
-          $parent->result_source->relationship_info($rel)->{cond},
+          $rels->{$rel}{cond},
           $child,
-          $parent,
+          $main_row,
+          $rel,
         );
 
         my @rows_to_add = ref $item->{$rel} eq 'ARRAY' ? @{$item->{$rel}} : ($item->{$rel});
@@ -1977,7 +2428,10 @@ sub _normalize_populate_args {
   my ($self, $arg) = @_;
 
   if (ref $arg eq 'ARRAY') {
-    if (ref $arg->[0] eq 'HASH') {
+    if (!@$arg) {
+      return [];
+    }
+    elsif (ref $arg->[0] eq 'HASH') {
       return $arg;
     }
     elsif (ref $arg->[0] eq 'ARRAY') {
@@ -2017,20 +2471,26 @@ sub pager {
   return $self->{pager} if $self->{pager};
 
   my $attrs = $self->{attrs};
-  $self->throw_exception("Can't create pager for non-paged rs")
-    unless $self->{attrs}{page};
+  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_count = (ref $self)->new($self->result_source, $count_attrs)->count;
 
-  return $self->{pager} = Data::Page->new(
-    $total_count,
+  my $total_rs = (ref $self)->new($self->result_source, $count_attrs);
+
+  require DBIx::Class::ResultSet::Pager;
+  return $self->{pager} = DBIx::Class::ResultSet::Pager->new(
+    sub { $total_rs->count },  #lazy-get the total
     $attrs->{rows},
-    $self->{attrs}{page}
+    $self->{attrs}{page},
   );
 }
 
@@ -2079,27 +2539,26 @@ sub new_result {
   $self->throw_exception( "new_result needs a hash" )
     unless (ref $values eq 'HASH');
 
-  my ($merged_cond, $cols_from_relations) = $self->_merge_cond_with_data($values);
+  my ($merged_cond, $cols_from_relations) = $self->_merge_with_rscond($values);
 
   my %new = (
     %$merged_cond,
     @$cols_from_relations
       ? (-cols_from_relations => $cols_from_relations)
       : (),
-    -source_handle => $self->_source_handle,
     -result_source => $self->result_source, # DO NOT REMOVE THIS, REQUIRED
   );
 
   return $self->result_class->new(\%new);
 }
 
-# _merge_cond_with_data
+# _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_cond_with_data {
+sub _merge_with_rscond {
   my ($self, $data) = @_;
 
   my (%new_data, @cols_from_relations);
@@ -2125,11 +2584,19 @@ sub _merge_cond_with_data {
     my %implied = %{$self->_remove_alias($collapsed_cond, $alias)};
 
     while ( my($col, $value) = each %implied ) {
-      if (ref($value) eq 'HASH' && keys(%$value) && (keys %$value)[0] eq '=') {
+      my $vref = ref $value;
+      if (
+        $vref eq 'HASH'
+          and
+        keys(%$value) == 1
+          and
+        (keys %$value)[0] eq '='
+      ) {
         $new_data{$col} = $value->{'='};
-        next;
       }
-      $new_data{$col} = $value if $self->_is_deterministic_value($value);
+      elsif( !$vref or $vref eq 'SCALAR' or blessed($value) ) {
+        $new_data{$col} = $value;
+      }
     }
   }
 
@@ -2141,20 +2608,6 @@ sub _merge_cond_with_data {
   return (\%new_data, \@cols_from_relations);
 }
 
-# _is_deterministic_value
-#
-# Make an effor to strip non-deterministic values from the condition,
-# to make sure new_result chokes less
-
-sub _is_deterministic_value {
-  my $self = shift;
-  my $value = shift;
-  my $ref_type = ref $value;
-  return 1 if $ref_type eq '' || $ref_type eq 'SCALAR';
-  return 1 if Scalar::Util::blessed($value);
-  return 0;
-}
-
 # _has_resolved_attr
 #
 # determines if the resultset defines at least one
@@ -2312,17 +2765,18 @@ sub as_query {
   $cd->cd_to_producer->find_or_new({ producer => $producer },
                                    { key => 'primary });
 
-Find an existing record from this resultset, based on its primary
-key, or a unique constraint. If none exists, instantiate a new result
-object and return it. The object will not be saved into your storage
-until you call L<DBIx::Class::Row/insert> on it.
+Find an existing record from this resultset using L</find>. if none exists,
+instantiate a new result object and return it. The object will not be saved
+into your storage until you call L<DBIx::Class::Row/insert> on it.
 
-You most likely want this method when looking for existing rows using
-a unique constraint that is not the primary key, or looking for
-related rows.
+You most likely want this method when looking for existing rows using a unique
+constraint that is not the primary key, or looking for related rows.
 
-If you want objects to be saved immediately, use L</find_or_create>
-instead.
+If you want objects to be saved immediately, use L</find_or_create> instead.
+
+B<Note>: Make sure to read the documentation of L</find> and understand the
+significance of the C<key> attribute, as its lack may skew your search, and
+subsequently result in spurious new objects.
 
 B<Note>: Take care when using C<find_or_new> with a table having
 columns with default values that you intend to be automatically
@@ -2398,7 +2852,7 @@ or C<has_one> resultset.  Note Arrayref.
   );
 
 Example of creating a new row and also creating a row in a related
-C<belongs_to>resultset. Note Hashref.
+C<belongs_to> resultset. Note Hashref.
 
   $cd_rs->create({
     title=>"Music for Silly Walks",
@@ -2464,6 +2918,10 @@ constraint. For example:
     { key => 'cd_artist_title' }
   );
 
+B<Note>: Make sure to read the documentation of L</find> and understand the
+significance of the C<key> attribute, as its lack may skew your search, and
+subsequently result in spurious row creation.
+
 B<Note>: Because find_or_create() reads from the database and then
 possibly inserts based on the result, this method is subject to a race
 condition. Another process could create a record in the table after
@@ -2479,6 +2937,23 @@ all in the call to C<find_or_create>, even when set to C<undef>.
 See also L</find> and L</update_or_create>. For information on how to declare
 unique constraints, see L<DBIx::Class::ResultSource/add_unique_constraint>.
 
+If you need to know if an existing row was found or a new one created use
+L</find_or_new> and L<DBIx::Class::Row/in_storage> instead. Don't forget
+to call L<DBIx::Class::Row/insert> to save the newly created row to the
+database!
+
+  my $cd = $schema->resultset('CD')->find_or_new({
+    cdid   => 5,
+    artist => 'Massive Attack',
+    title  => 'Mezzanine',
+    year   => 2005,
+  });
+
+  if( $cd->in_storage ) {
+      # do some stuff
+      $cd->insert;
+  }
+
 =cut
 
 sub find_or_create {
@@ -2497,16 +2972,15 @@ sub find_or_create {
 
 =item Arguments: \%col_values, { key => $unique_constraint }?
 
-=item Return Value: $rowobject
+=item Return Value: $row_object
 
 =back
 
   $resultset->update_or_create({ col => $val, ... });
 
-First, searches for an existing row matching one of the unique constraints
-(including the primary key) on the source of this resultset. If a row is
-found, updates it with the other given column values. Otherwise, creates a new
-row.
+Like L</find_or_create>, but if a row is found it is immediately updated via
+C<< $found_row->update (\%col_values) >>.
+
 
 Takes an optional C<key> attribute to search on a specific unique constraint.
 For example:
@@ -2525,17 +2999,12 @@ For example:
     producer => $producer,
     name => 'harry',
   }, {
-    key => 'primary,
+    key => 'primary',
   });
 
-
-If no C<key> is specified, it searches on all unique constraints defined on the
-source, including the primary key.
-
-If the C<key> is specified as C<primary>, it searches only on the primary key.
-
-See also L</find> and L</find_or_create>. For information on how to declare
-unique constraints, see L<DBIx::Class::ResultSource/add_unique_constraint>.
+B<Note>: Make sure to read the documentation of L</find> and understand the
+significance of the C<key> attribute, as its lack may skew your search, and
+subsequently result in spurious row creation.
 
 B<Note>: Take care when using C<update_or_create> with a table having
 columns with default values that you intend to be automatically
@@ -2543,6 +3012,28 @@ supplied by the database (e.g. an auto_increment primary key column).
 In normal usage, the value of such columns should NOT be included at
 all in the call to C<update_or_create>, even when set to C<undef>.
 
+See also L</find> and L</find_or_create>. For information on how to declare
+unique constraints, see L<DBIx::Class::ResultSource/add_unique_constraint>.
+
+If you need to know if an existing row was updated or a new one created use
+L</update_or_new> and L<DBIx::Class::Row/in_storage> instead. Don't forget
+to call L<DBIx::Class::Row/insert> to save the newly created row to the
+database!
+
+  my $cd = $schema->resultset('CD')->update_or_new(
+    {
+      artist => 'Massive Attack',
+      title  => 'Mezzanine',
+      year   => 1998,
+    },
+    { key => 'cd_artist_title' }
+  );
+
+  if( $cd->in_storage ) {
+      # do some stuff
+      $cd->insert;
+  }
+
 =cut
 
 sub update_or_create {
@@ -2571,13 +3062,9 @@ sub update_or_create {
 
   $resultset->update_or_new({ col => $val, ... });
 
-First, searches for an existing row matching one of the unique constraints
-(including the primary key) on the source of this resultset. If a row is
-found, updates it with the other given column values. Otherwise, instantiate
-a new result object and return it. The object will not be saved into your storage
-until you call L<DBIx::Class::Row/insert> on it.
+Like L</find_or_new> but if a row is found it is immediately updated via
+C<< $found_row->update (\%col_values) >>.
 
-Takes an optional C<key> attribute to search on a specific unique constraint.
 For example:
 
   # In your application
@@ -2598,6 +3085,10 @@ For example:
       $cd->insert;
   }
 
+B<Note>: Make sure to read the documentation of L</find> and understand the
+significance of the C<key> attribute, as its lack may skew your search, and
+subsequently result in spurious new objects.
+
 B<Note>: Take care when using C<update_or_new> with a table having
 columns with default values that you intend to be automatically
 supplied by the database (e.g. an auto_increment primary key column).
@@ -2628,7 +3119,7 @@ sub update_or_new {
 
 =item Arguments: none
 
-=item Return Value: \@cache_objects?
+=item Return Value: \@cache_objects | undef
 
 =back
 
@@ -2676,7 +3167,7 @@ sub set_cache {
 
 =item Arguments: none
 
-=item Return Value: []
+=item Return Value: undef
 
 =back
 
@@ -2719,7 +3210,7 @@ sub is_paged {
 
 sub is_ordered {
   my ($self) = @_;
-  return scalar $self->result_source->storage->_parse_order_by($self->{attrs}{order_by});
+  return scalar $self->result_source->storage->_extract_order_criteria($self->{attrs}{order_by});
 }
 
 =head2 related_resultset
@@ -2762,7 +3253,7 @@ sub related_resultset {
     # (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->_straight_join_to_node ($attrs->{from}, $alias);
+    $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
@@ -2832,9 +3323,9 @@ source alias of the current result set:
 
     my $me = $self->current_source_alias;
 
-    return $self->search(
+    return $self->search({
       "$me.modified" => $user->id,
-    );
+    });
   }
 
 =cut
@@ -2895,16 +3386,26 @@ but because we isolated the group by into a subselect the above works.
 =cut
 
 sub as_subselect_rs {
-   my $self = shift;
+  my $self = shift;
+
+  my $attrs = $self->_resolved_attrs;
 
-   return $self->result_source->resultset->search( undef, {
-      alias => $self->current_source_alias,
-      from => [{
-            $self->current_source_alias => $self->as_query,
-            -alias         => $self->current_source_alias,
-            -source_handle => $self->result_source->handle,
-         }]
-   });
+  my $fresh_rs = (ref $self)->new (
+    $self->result_source
+  );
+
+  # these pieces will be locked in the subquery
+  delete $fresh_rs->{cond};
+  delete @{$fresh_rs->{attrs}}{qw/where bind/};
+
+  return $fresh_rs->search( {}, {
+    from => [{
+      $attrs->{alias} => $self->as_query,
+      -alias  => $attrs->{alias},
+      -rsrc   => $self->result_source,
+    }],
+    alias => $attrs->{alias},
+  });
 }
 
 # This code is called by search_related, and makes sure there
@@ -2927,9 +3428,9 @@ sub _chain_relationship {
 
   # we need to take the prefetch the attrs into account before we
   # ->_resolve_join as otherwise they get lost - captainL
-  my $join = $self->_merge_attr( $attrs->{join}, $attrs->{prefetch} );
+  my $join = $self->_merge_joinpref_attr( $attrs->{join}, $attrs->{prefetch} );
 
-  delete @{$attrs}{qw/join prefetch collapse distinct select as columns +select +as +columns/};
+  delete @{$attrs}{qw/join prefetch collapse group_by distinct select as columns +select +as +columns/};
 
   my $seen = { %{ (delete $attrs->{seen_join}) || {} } };
 
@@ -2945,17 +3446,17 @@ sub _chain_relationship {
     # are resolved (prefetch is useless - we are wrapping
     # a subquery anyway).
     my $rs_copy = $self->search;
-    $rs_copy->{attrs}{join} = $self->_merge_attr (
+    $rs_copy->{attrs}{join} = $self->_merge_joinpref_attr (
       $rs_copy->{attrs}{join},
       delete $rs_copy->{attrs}{prefetch},
     );
 
     $from = [{
-      -source_handle => $source->handle,
-      -alias => $attrs->{alias},
+      -rsrc   => $source,
+      -alias  => $attrs->{alias},
       $attrs->{alias} => $rs_copy->as_query,
     }];
-    delete @{$attrs}{@force_subq_attrs, 'where'};
+    delete @{$attrs}{@force_subq_attrs, qw/where bind/};
     $seen->{-relation_chain_depth} = 0;
   }
   elsif ($attrs->{from}) {  #shallow copy suffices
@@ -2963,7 +3464,7 @@ sub _chain_relationship {
   }
   else {
     $from = [{
-      -source_handle => $source->handle,
+      -rsrc  => $source,
       -alias => $attrs->{alias},
       $attrs->{alias} => $source->from,
     }];
@@ -3028,100 +3529,79 @@ sub _resolved_attrs {
   my $source = $self->result_source;
   my $alias  = $attrs->{alias};
 
-  $attrs->{columns} ||= delete $attrs->{cols} if exists $attrs->{cols};
-  my @colbits;
-
-  # build columns (as long as select isn't set) into a set of as/select hashes
-  unless ( $attrs->{select} ) {
-
-    my @cols;
-    if ( ref $attrs->{columns} eq 'ARRAY' ) {
-      @cols = @{ delete $attrs->{columns}}
-    } elsif ( defined $attrs->{columns} ) {
-      @cols = delete $attrs->{columns}
-    } else {
-      @cols = $source->columns
-    }
+  # default selection list
+  $attrs->{columns} = [ $source->columns ]
+    unless List::Util::first { exists $attrs->{$_} } qw/columns cols select as/;
 
-    for (@cols) {
-      if ( ref $_ eq 'HASH' ) {
-        push @colbits, $_
-      } else {
-        my $key = /^\Q${alias}.\E(.+)$/
-          ? "$1"
-          : "$_";
-        my $value = /\./
-          ? "$_"
-          : "${alias}.$_";
-        push @colbits, { $key => $value };
-      }
-    }
+  # merge selectors together
+  for (qw/columns select as/) {
+    $attrs->{$_} = $self->_merge_attr($attrs->{$_}, delete $attrs->{"+$_"})
+      if $attrs->{$_} or $attrs->{"+$_"};
   }
 
-  # add the additional columns on
-  foreach (qw{include_columns +columns}) {
-    if ( $attrs->{$_} ) {
-      my @list = ( ref($attrs->{$_}) eq 'ARRAY' )
-        ? @{ delete $attrs->{$_} }
-        : delete $attrs->{$_};
-      for (@list) {
-        if ( ref($_) eq 'HASH' ) {
-          push @colbits, $_
-        } else {
-          my $key = ( split /\./, $_ )[-1];
-          my $value = ( /\./ ? $_ : "$alias.$_" );
-          push @colbits, { $key => $value };
+  # disassemble columns
+  my (@sel, @as);
+  if (my $cols = delete $attrs->{columns}) {
+    for my $c (ref $cols eq 'ARRAY' ? @$cols : $cols) {
+      if (ref $c eq 'HASH') {
+        for my $as (keys %$c) {
+          push @sel, $c->{$as};
+          push @as, $as;
         }
       }
+      else {
+        push @sel, $c;
+        push @as, $c;
+      }
     }
   }
 
-  # start with initial select items
-  if ( $attrs->{select} ) {
-    $attrs->{select} =
-        ( ref $attrs->{select} eq 'ARRAY' )
-      ? [ @{ $attrs->{select} } ]
-      : [ $attrs->{select} ];
+  # when trying to weed off duplicates later do not go past this point -
+  # everything added from here on is unbalanced "anyone's guess" stuff
+  my $dedup_stop_idx = $#as;
 
-    if ( $attrs->{as} ) {
-      $attrs->{as} =
-        (
-          ref $attrs->{as} eq 'ARRAY'
-            ? [ @{ $attrs->{as} } ]
-            : [ $attrs->{as} ]
-        )
-    } else {
-      $attrs->{as} = [ map {
-         m/^\Q${alias}.\E(.+)$/
-           ? $1
-           : $_
-         } @{ $attrs->{select} }
-      ]
-    }
-  }
-  else {
-    # otherwise we intialise select & as to empty
-    $attrs->{select} = [];
-    $attrs->{as}     = [];
-  }
+  push @as, @{ ref $attrs->{as} eq 'ARRAY' ? $attrs->{as} : [ $attrs->{as} ] }
+    if $attrs->{as};
+  push @sel, @{ ref $attrs->{select} eq 'ARRAY' ? $attrs->{select} : [ $attrs->{select} ] }
+    if $attrs->{select};
 
-  # now add colbits to select/as
-  push @{ $attrs->{select} }, map values %{$_}, @colbits;
-  push @{ $attrs->{as}     }, map keys   %{$_}, @colbits;
+  # assume all unqualified selectors to apply to the current alias (legacy stuff)
+  for (@sel) {
+    $_ = (ref $_ or $_ =~ /\./) ? $_ : "$alias.$_";
+  }
 
-  if ( my $adds = delete $attrs->{'+select'} ) {
-    $adds = [$adds] unless ref $adds eq 'ARRAY';
-    push @{ $attrs->{select} },
-      map { /\./ || ref $_ ? $_ : "$alias.$_" } @$adds;
+  # disqualify all $alias.col as-bits (collapser mandated)
+  for (@as) {
+    $_ = ($_ =~ /^\Q$alias.\E(.+)$/) ? $1 : $_;
   }
-  if ( my $adds = delete $attrs->{'+as'} ) {
-    $adds = [$adds] unless ref $adds eq 'ARRAY';
-    push @{ $attrs->{as} }, @$adds;
+
+  # de-duplicate the result (remove *identical* select/as pairs)
+  # and also die on duplicate {as} pointing to different {select}s
+  # not using a c-style for as the condition is prone to shrinkage
+  my $seen;
+  my $i = 0;
+  while ($i <= $dedup_stop_idx) {
+    if ($seen->{"$sel[$i] \x00\x00 $as[$i]"}++) {
+      splice @sel, $i, 1;
+      splice @as, $i, 1;
+      $dedup_stop_idx--;
+    }
+    elsif ($seen->{$as[$i]}++) {
+      $self->throw_exception(
+        "inflate_result() alias '$as[$i]' specified twice with different SQL-side {select}-ors"
+      );
+    }
+    else {
+      $i++;
+    }
   }
 
+  $attrs->{select} = \@sel;
+  $attrs->{as} = \@as;
+
   $attrs->{from} ||= [{
-    -source_handle => $source->handle,
-    -alias => $self->{attrs}{alias},
+    -rsrc   => $source,
+    -alias  => $self->{attrs}{alias},
     $self->{attrs}{alias} => $source->from,
   }];
 
@@ -3130,10 +3610,10 @@ sub _resolved_attrs {
     $self->throw_exception ('join/prefetch can not be used with a custom {from}')
       if ref $attrs->{from} ne 'ARRAY';
 
-    my $join = delete $attrs->{join} || {};
+    my $join = (delete $attrs->{join}) || {};
 
     if ( defined $attrs->{prefetch} ) {
-      $join = $self->_merge_attr( $join, $attrs->{prefetch} );
+      $join = $self->_merge_joinpref_attr( $join, $attrs->{prefetch} );
     }
 
     $attrs->{from} =    # have to copy here to avoid corrupting the original
@@ -3167,30 +3647,29 @@ sub _resolved_attrs {
   # subquery (since a group_by is present)
   if (delete $attrs->{distinct}) {
     if ($attrs->{group_by}) {
-      carp ("Useless use of distinct on a grouped resultset ('distinct' is ignored when a 'group_by' is present)");
+      carp_unique ("Useless use of distinct on a grouped resultset ('distinct' is ignored when a 'group_by' is present)");
     }
     else {
-      $attrs->{group_by} = [ grep { !ref($_) || (ref($_) ne 'HASH') } @{$attrs->{select}} ];
-
-      # add any order_by parts that are not already present in the group_by
-      # we need to be careful not to add any named functions/aggregates
-      # i.e. select => [ ... { count => 'foo', -as 'foocount' } ... ]
-      my %already_grouped = map { $_ => 1 } (@{$attrs->{group_by}});
-
-      my $storage = $self->result_source->schema->storage;
-
-      my $rs_column_list = $storage->_resolve_column_info ($attrs->{from});
-
-      for my $chunk ($storage->_parse_order_by($attrs->{order_by})) {
-        if ($rs_column_list->{$chunk} && not $already_grouped{$chunk}++) {
-          push @{$attrs->{group_by}}, $chunk;
-        }
-      }
+      # distinct affects only the main selection part, not what prefetch may
+      # add below.
+      $attrs->{group_by} = $source->storage->_group_over_selection (
+        $attrs->{from},
+        $attrs->{select},
+        $attrs->{order_by},
+      );
     }
   }
 
   # generate selections based on the prefetch helper
-  if ( my $prefetch = delete $attrs->{prefetch} ) {
+  my $prefetch;
+  $prefetch = $self->_merge_joinpref_attr( {}, delete $attrs->{prefetch} )
+    if defined $attrs->{prefetch};
+
+  if ($prefetch) {
+
+    $self->throw_exception("Unable to prefetch, resultset contains an unnamed selector $attrs->{_dark_selector}{string}")
+      if $attrs->{_dark_selector};
+
     $attrs->{collapse} = 1;
 
     # this is a separate structure (we don't look in {from} directly)
@@ -3217,9 +3696,12 @@ sub _resolved_attrs {
     my @prefetch = $source->_resolve_prefetch( $prefetch, $alias, $join_map );
 
     # we need to somehow mark which columns came from prefetch
-    $attrs->{_prefetch_select} = [ map { $_->[0] } @prefetch ];
+    if (@prefetch) {
+      my $sel_end = $#{$attrs->{select}};
+      $attrs->{_prefetch_selector_range} = [ $sel_end + 1, $sel_end + @prefetch ];
+    }
 
-    push @{ $attrs->{select} }, @{$attrs->{_prefetch_select}};
+    push @{ $attrs->{select} }, (map { $_->[0] } @prefetch);
     push @{ $attrs->{as} }, (map { $_->[1] } @prefetch);
   }
 
@@ -3248,6 +3730,7 @@ sub _resolved_attrs {
     }
   }
 
+
   # if both page and offset are specified, produce a combined offset
   # even though it doesn't make much sense, this is what pre 081xx has
   # been doing
@@ -3333,7 +3816,7 @@ sub _calculate_score {
   }
 }
 
-sub _merge_attr {
+sub _merge_joinpref_attr {
   my ($self, $orig, $import) = @_;
 
   return $import unless defined($orig);
@@ -3355,6 +3838,7 @@ sub _merge_attr {
       $position++;
     }
     my ($import_key) = ( ref $import_element eq 'HASH' ) ? keys %{$import_element} : ($import_element);
+    $import_key = '' if not defined $import_key;
 
     if ($best_candidate->{score} == 0 || exists $seen_keys->{$import_key}) {
       push( @{$orig}, $import_element );
@@ -3365,7 +3849,7 @@ sub _merge_attr {
         $orig->[$best_candidate->{position}] = $import_element;
       } elsif (ref $import_element eq 'HASH') {
         my ($key) = keys %{$orig_best};
-        $orig->[$best_candidate->{position}] = { $key => $self->_merge_attr($orig_best->{$key}, $import_element->{$key}) };
+        $orig->[$best_candidate->{position}] = { $key => $self->_merge_joinpref_attr($orig_best->{$key}, $import_element->{$key}) };
       }
     }
     $seen_keys->{$import_key} = 1; # don't merge the same key twice
@@ -3374,16 +3858,119 @@ sub _merge_attr {
   return $orig;
 }
 
-sub result_source {
-    my $self = shift;
+{
+  my $hm;
 
-    if (@_) {
-        $self->_source_handle($_[0]->handle);
-    } else {
-        $self->_source_handle->resolve;
-    }
+  sub _merge_attr {
+    $hm ||= do {
+      require Hash::Merge;
+      my $hm = Hash::Merge->new;
+
+      $hm->specify_behavior({
+        SCALAR => {
+          SCALAR => sub {
+            my ($defl, $defr) = map { defined $_ } (@_[0,1]);
+
+            if ($defl xor $defr) {
+              return [ $defl ? $_[0] : $_[1] ];
+            }
+            elsif (! $defl) {
+              return [];
+            }
+            elsif (__HM_DEDUP and $_[0] eq $_[1]) {
+              return [ $_[0] ];
+            }
+            else {
+              return [$_[0], $_[1]];
+            }
+          },
+          ARRAY => sub {
+            return $_[1] if !defined $_[0];
+            return $_[1] if __HM_DEDUP and List::Util::first { $_ eq $_[0] } @{$_[1]};
+            return [$_[0], @{$_[1]}]
+          },
+          HASH  => sub {
+            return [] if !defined $_[0] and !keys %{$_[1]};
+            return [ $_[1] ] if !defined $_[0];
+            return [ $_[0] ] if !keys %{$_[1]};
+            return [$_[0], $_[1]]
+          },
+        },
+        ARRAY => {
+          SCALAR => sub {
+            return $_[0] if !defined $_[1];
+            return $_[0] if __HM_DEDUP and List::Util::first { $_ eq $_[1] } @{$_[0]};
+            return [@{$_[0]}, $_[1]]
+          },
+          ARRAY => sub {
+            my @ret = @{$_[0]} or return $_[1];
+            return [ @ret, @{$_[1]} ] unless __HM_DEDUP;
+            my %idx = map { $_ => 1 } @ret;
+            push @ret, grep { ! defined $idx{$_} } (@{$_[1]});
+            \@ret;
+          },
+          HASH => sub {
+            return [ $_[1] ] if ! @{$_[0]};
+            return $_[0] if !keys %{$_[1]};
+            return $_[0] if __HM_DEDUP and List::Util::first { $_ eq $_[1] } @{$_[0]};
+            return [ @{$_[0]}, $_[1] ];
+          },
+        },
+        HASH => {
+          SCALAR => sub {
+            return [] if !keys %{$_[0]} and !defined $_[1];
+            return [ $_[0] ] if !defined $_[1];
+            return [ $_[1] ] if !keys %{$_[0]};
+            return [$_[0], $_[1]]
+          },
+          ARRAY => sub {
+            return [] if !keys %{$_[0]} and !@{$_[1]};
+            return [ $_[0] ] if !@{$_[1]};
+            return $_[1] if !keys %{$_[0]};
+            return $_[1] if __HM_DEDUP and List::Util::first { $_ eq $_[0] } @{$_[1]};
+            return [ $_[0], @{$_[1]} ];
+          },
+          HASH => sub {
+            return [] if !keys %{$_[0]} and !keys %{$_[1]};
+            return [ $_[0] ] if !keys %{$_[1]};
+            return [ $_[1] ] if !keys %{$_[0]};
+            return [ $_[0] ] if $_[0] eq $_[1];
+            return [ $_[0], $_[1] ];
+          },
+        }
+      } => 'DBIC_RS_ATTR_MERGER');
+      $hm;
+    };
+
+    return $hm->merge ($_[1], $_[2]);
+  }
+}
+
+sub STORABLE_freeze {
+  my ($self, $cloning) = @_;
+  my $to_serialize = { %$self };
+
+  # A cursor in progress can't be serialized (and would make little sense anyway)
+  delete $to_serialize->{cursor};
+
+  # nor is it sensical to store a not-yet-fired-count pager
+  if ($to_serialize->{pager} and ref $to_serialize->{pager}{total_entries} eq 'CODE') {
+    delete $to_serialize->{pager};
+  }
+
+  Storable::nfreeze($to_serialize);
+}
+
+# need this hook for symmetry
+sub STORABLE_thaw {
+  my ($self, $cloning, $serialized) = @_;
+
+  %$self = %{ Storable::thaw($serialized) };
+
+  $self;
 }
 
+
 =head2 throw_exception
 
 See L<DBIx::Class::Schema/throw_exception> for details.
@@ -3393,8 +3980,8 @@ See L<DBIx::Class::Schema/throw_exception> for details.
 sub throw_exception {
   my $self=shift;
 
-  if (ref $self && $self->_source_handle->schema) {
-    $self->_source_handle->schema->throw_exception(@_)
+  if (ref $self and my $rsrc = $self->result_source) {
+    $rsrc->throw_exception(@_)
   }
   else {
     DBIx::Class::Exception->throw(@_);
@@ -3458,6 +4045,15 @@ it and sets C<select> from that, then auto-populates C<as> from
 C<select> as normal. (You may also use the C<cols> attribute, as in
 earlier versions of DBIC.)
 
+Essentially C<columns> does the same as L</select> and L</as>.
+
+    columns => [ 'foo', { bar => 'baz' } ]
+
+is the same as
+
+    select => [qw/foo baz/],
+    as => [qw/foo bar/]
+
 =head2 +columns
 
 =over 4
@@ -3481,6 +4077,10 @@ passed to object inflation. Note that the 'artist' is the name of the
 column (or relationship) accessor, and 'name' is the name of the column
 accessor in the related table.
 
+B<NOTE:> You need to explicitly quote '+columns' when defining the attribute.
+Not doing so causes Perl to incorrectly interpret +columns as a bareword with a
+unary plus operator before it.
+
 =head2 include_columns
 
 =over 4
@@ -3507,23 +4107,31 @@ names:
     select => [
       'name',
       { count => 'employeeid' },
-      { sum => 'salary' }
+      { max => { length => 'name' }, -as => 'longest_name' }
     ]
   });
 
-When you use function/stored procedure names and do not supply an C<as>
-attribute, the column names returned are storage-dependent. E.g. MySQL would
-return a column named C<count(employeeid)> in the above example.
+  # Equivalent SQL
+  SELECT name, COUNT( employeeid ), MAX( LENGTH( name ) ) AS longest_name FROM employee
 
-B<NOTE:> You will almost always need a corresponding 'as' entry when you use
-'select'.
+B<NOTE:> You will almost always need a corresponding L</as> attribute when you
+use L</select>, to instruct DBIx::Class how to store the result of the column.
+Also note that the L</as> attribute has nothing to do with the SQL-side 'AS'
+identifier aliasing. You can however alias a function, so you can use it in
+e.g. an C<ORDER BY> clause. This is done via the C<-as> B<select function
+attribute> supplied as shown in the example above.
+
+B<NOTE:> You need to explicitly quote '+select'/'+as' when defining the attributes.
+Not doing so causes Perl to incorrectly interpret them as a bareword with a
+unary plus operator before it.
 
 =head2 +select
 
 =over 4
 
 Indicates additional columns to be selected from storage.  Works the same as
-L</select> but adds columns to the selection.
+L</select> but adds columns to the default selection, instead of specifying
+an explicit list.
 
 =back
 
@@ -3543,25 +4151,26 @@ Indicates additional column names for those added via L</+select>. See L</as>.
 
 =back
 
-Indicates column names for object inflation. That is, C<as>
-indicates the name that the column can be accessed as via the
-C<get_column> method (or via the object accessor, B<if one already
-exists>).  It has nothing to do with the SQL code C<SELECT foo AS bar>.
-
-The C<as> attribute is used in conjunction with C<select>,
-usually when C<select> contains one or more function or stored
-procedure names:
+Indicates column names for object inflation. That is L</as> indicates the
+slot name in which the column value will be stored within the
+L<Row|DBIx::Class::Row> object. The value will then be accessible via this
+identifier by the C<get_column> method (or via the object accessor B<if one
+with the same name already exists>) as shown below. The L</as> attribute has
+B<nothing to do> with the SQL-side C<AS>. See L</select> for details.
 
   $rs = $schema->resultset('Employee')->search(undef, {
     select => [
       'name',
-      { count => 'employeeid' }
+      { count => 'employeeid' },
+      { max => { length => 'name' }, -as => 'longest_name' }
     ],
-    as => ['name', 'employee_count'],
+    as => [qw/
+      name
+      employee_count
+      max_name_length
+    /],
   });
 
-  my $employee = $rs->first(); # get the first Employee
-
 If the object against which the search is performed already has an accessor
 matching a column name specified in C<as>, the value can be retrieved using
 the accessor as normal:
@@ -3576,16 +4185,6 @@ use C<get_column> instead:
 You can create your own accessors if required - see
 L<DBIx::Class::Manual::Cookbook> for details.
 
-Please note: This will NOT insert an C<AS employee_count> into the SQL
-statement produced, it is used for internal access only. Thus
-attempting to use the accessor in an C<order_by> clause or similar
-will fail miserably.
-
-To get around this limitation, you can supply literal SQL to your
-C<select> attribute that contains the C<AS alias> text, e.g.
-
-  select => [\'myfield AS alias']
-
 =head2 join
 
 =over 4
@@ -3690,28 +4289,122 @@ case.
 Simple prefetches will be joined automatically, so there is no need
 for a C<join> attribute in the above search.
 
-C<prefetch> can be used with the following relationship types: C<belongs_to>,
-C<has_one> (or if you're using C<add_relationship>, any relationship declared
-with an accessor type of 'single' or 'filter'). A more complex example that
-prefetches an artists cds, the tracks on those cds, and the tags associated
-with that artist is given below (assuming many-to-many from artists to tags):
+L</prefetch> can be used with the any of the relationship types and
+multiple prefetches can be specified together. Below is a more complex
+example that prefetches a CD's artist, its liner notes (if present),
+the cover image, the tracks on that cd, and the guests on those
+tracks.
 
- my $rs = $schema->resultset('Artist')->search(
+ # 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
+     ]
+   }
+ );
+
+This will produce SQL like the following:
+
+ 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
+
+Now the C<artist>, C<record_label>, C<liner_note>, C<cover_image>,
+C<tracks>, and C<guests> of the CD will all be available through the
+relationship accessors without the need for additional queries to the
+database.
+
+However, there is one caveat to be observed: it can be dangerous to
+prefetch more than one L<has_many|DBIx::Class::Relationship/has_many>
+relationship on a given level. e.g.:
+
+ my $rs = $schema->resultset('CD')->search(
    undef,
    {
      prefetch => [
-       { cds => 'tracks' },
-       { artist_tags => 'tags' }
+       'tracks',                         # has_many
+       { cd_to_producer => 'producer' }, # has_many => belongs_to (i.e. m2m)
      ]
    }
  );
 
+In fact, C<DBIx::Class> will emit the following warning:
 
-B<NOTE:> If you specify a C<prefetch> attribute, the C<join> and C<select>
-attributes will be ignored.
+ 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.
 
-B<CAVEATs>: Prefetch does a lot of deep magic. As such, it may not behave
-exactly as you might expect.
+The collapser currently can't identify duplicate tuples for multiple
+L<has_many|DBIx::Class::Relationship/has_many> relationships and as a
+result the second L<has_many|DBIx::Class::Relationship/has_many>
+relation could contain redundant objects.
+
+=head3 Using L</prefetch> with L</join>
+
+L</prefetch> implies a L</join> with the equivalent argument, and is
+properly merged with any existing L</join> specification. So the
+following:
+
+  my $rs = $schema->resultset('CD')->search(
+   {'record_label.name' => 'Music Product Ltd.'},
+   {
+     join     => {artist => 'record_label'},
+     prefetch => 'artist',
+   }
+ );
+
+... will work, searching on the record label's name, but only
+prefetching the C<artist>.
+
+=head3 Using L</prefetch> with L</select> / L</+select> / L</as> / L</+as>
+
+L</prefetch> implies a L</+select>/L</+as> with the fields of the
+prefetched relations.  So given:
+
+  my $rs = $schema->resultset('CD')->search(
+   undef,
+   {
+     select   => ['cd.title'],
+     as       => ['cd_title'],
+     prefetch => 'artist',
+   }
+ );
+
+The L</select> becomes: C<'cd.title', 'artist.*'> and the L</as>
+becomes: C<'cd_title', 'artist.*'>.
+
+=head3 CAVEATS
+
+Prefetch does a lot of deep magic. As such, it may not behave exactly
+as you might expect.
 
 =over 4
 
@@ -3758,7 +4451,7 @@ Makes the resultset paged and specifies the page to retrieve. Effectively
 identical to creating a non-pages resultset and then calling ->page($page)
 on it.
 
-If L<rows> attribute is not specified it defaults to 10 rows per page.
+If L</rows> attribute is not specified it defaults to 10 rows per page.
 
 When you have a paged resultset, L</count> will only return the number
 of rows in the page. To get the total, use the L</pager> and call
@@ -3786,6 +4479,24 @@ rows per page if the page attribute or method is used.
 Specifies the (zero-based) row number for the  first row to be returned, or the
 of the first row of the first page if paging is used.
 
+=head2 software_limit
+
+=over 4
+
+=item Value: (0 | 1)
+
+=back
+
+When combined with L</rows> and/or L</offset> the generated SQL will not
+include any limit dialect stanzas. Instead the entire result will be selected
+as if no limits were specified, and DBIC will perform the limit locally, by
+artificially advancing and finishing the resulting L</cursor>.
+
+This is the recommended way of performing resultset limiting when no sane RDBMS
+implementation is available (e.g.
+L<Sybase ASE|DBIx::Class::Storage::DBI::Sybase::ASE> using the
+L<Generic Sub Query|DBIx::Class::SQLMaker::LimitDialects/GenericSubQ> hack)
+
 =head2 group_by
 
 =over 4
@@ -3810,7 +4521,11 @@ HAVING is a select statement attribute that is applied between GROUP BY and
 ORDER BY. It is applied to the after the grouping calculations have been
 done.
 
-  having => { 'count(employee)' => { '>=', 100 } }
+  having => { 'count_employee' => { '>=', 100 } }
+
+or with an in-place function in which case literal SQL is required:
+
+  having => \[ 'count(employee) >= ?', [ count => 100 ] ]
 
 =head2 distinct
 
@@ -3835,6 +4550,8 @@ Adds to the WHERE clause.
 Can be overridden by passing C<< { where => undef } >> as an attribute
 to a resultset.
 
+For more complicated where clauses see L<SQL::Abstract/WHERE CLAUSES>.
+
 =back
 
 =head2 cache