X-Git-Url: http://git.shadowcat.co.uk/gitweb/gitweb.cgi?a=blobdiff_plain;f=lib%2FDBIx%2FClass%2FResultSet.pm;h=8745c5f796eff325244ccda8f9fe62573eb56e89;hb=30681c23187d5f13d57eda0c97dda1be5fb291d1;hp=91ae0bb334e9791d1de91036ef040c636fc1d766;hpb=31073ac79e22ae0f977c7e0fc4e9857d250143c4;p=dbsrgits%2FDBIx-Class.git

diff --git a/lib/DBIx/Class/ResultSet.pm b/lib/DBIx/Class/ResultSet.pm
index 91ae0bb..8745c5f 100644
--- a/lib/DBIx/Class/ResultSet.pm
+++ b/lib/DBIx/Class/ResultSet.pm
@@ -4,9 +4,11 @@ use strict;
 use warnings;
 use base qw/DBIx::Class/;
 use DBIx::Class::Carp;
-use DBIx::Class::Exception;
 use DBIx::Class::ResultSetColumn;
-use Scalar::Util qw/blessed weaken/;
+use Scalar::Util qw/blessed weaken reftype/;
+use DBIx::Class::_Util qw(
+  fail_on_internal_wantarray fail_on_internal_call UNRESOLVABLE_CONDITION
+);
 use Try::Tiny;
 use Data::Compare (); # no imports!!! guard against insane architecture
 
@@ -26,6 +28,10 @@ use overload
         'bool'   => "_bool",
         fallback => 1;
 
+# this is real - CDBICompat overrides it with insanity
+# yes, prototype won't matter, but that's for now ;)
+sub _bool () { 1 }
+
 __PACKAGE__->mk_group_accessors('simple' => qw/_result_class result_source/);
 
 =head1 NAME
@@ -74,34 +80,6 @@ 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
 
 =head2 Chaining resultsets
@@ -138,11 +116,15 @@ another.
 
 =head3 Resolving conditions and attributes
 
-When a resultset is chained from another resultset, conditions and
-attributes with the same keys need resolving.
+When a resultset is chained from another resultset (e.g.:
+C<< my $new_rs = $old_rs->search(\%extra_cond, \%attrs) >>), conditions
+and attributes with the same keys need resolving.
+
+If any of L</columns>, L</select>, L</as> are present, they reset the
+original selection, and start the selection "clean".
 
-L</join>, L</prefetch>, L</+select>, L</+as> attributes are merged
-into the existing ones from the original resultset.
+The L</join>, L</prefetch>, L</+columns>, L</+select>, L</+as> attributes
+are merged into the existing ones from the original resultset.
 
 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
@@ -185,6 +167,93 @@ Which is the same as:
 
 See: L</search>, L</count>, L</get_column>, L</all>, L</create>.
 
+=head2 Custom ResultSet classes
+
+To add methods to your resultsets, you can subclass L<DBIx::Class::ResultSet>, similar to:
+
+  package MyApp::Schema::ResultSet::User;
+
+  use strict;
+  use warnings;
+
+  use base 'DBIx::Class::ResultSet';
+
+  sub active {
+    my $self = shift;
+    $self->search({ $self->current_source_alias . '.active' => 1 });
+  }
+
+  sub unverified {
+    my $self = shift;
+    $self->search({ $self->current_source_alias . '.verified' => 0 });
+  }
+
+  sub created_n_days_ago {
+    my ($self, $days_ago) = @_;
+    $self->search({
+      $self->current_source_alias . '.create_date' => {
+        '<=',
+      $self->result_source->schema->storage->datetime_parser->format_datetime(
+        DateTime->now( time_zone => 'UTC' )->subtract( days => $days_ago )
+      )}
+    });
+  }
+
+  sub users_to_warn { shift->active->unverified->created_n_days_ago(7) }
+
+  1;
+
+See L<DBIx::Class::Schema/load_namespaces> on how DBIC can discover and
+automatically attach L<Result|DBIx::Class::Manual::ResultClass>-specific
+L<ResulSet|DBIx::Class::ResultSet> classes.
+
+=head3 ResultSet subclassing with Moose and similar constructor-providers
+
+Using L<Moose> or L<Moo> in your ResultSet classes is usually overkill, but
+you may find it useful if your ResultSets contain a lot of business logic
+(e.g. C<has xml_parser>, C<has json>, etc) or if you just prefer to organize
+your code via roles.
+
+In order to write custom ResultSet classes with L<Moo> you need to use the
+following template. The L<BUILDARGS|Moo/BUILDARGS> is necessary due to the
+unusual signature of the L<constructor provided by DBIC
+|DBIx::Class::ResultSet/new> C<< ->new($source, \%args) >>.
+
+  use Moo;
+  extends 'DBIx::Class::ResultSet';
+  sub BUILDARGS { $_[2] } # ::RS::new() expects my ($class, $rsrc, $args) = @_
+
+  ...your code...
+
+  1;
+
+If you want to build your custom ResultSet classes with L<Moose>, you need
+a similar, though a little more elaborate template in order to interface the
+inlining of the L<Moose>-provided
+L<object constructor|Moose::Manual::Construction/WHERE'S THE CONSTRUCTOR?>,
+with the DBIC one.
+
+  package MyApp::Schema::ResultSet::User;
+
+  use Moose;
+  use MooseX::NonMoose;
+  extends 'DBIx::Class::ResultSet';
+
+  sub BUILDARGS { $_[2] } # ::RS::new() expects my ($class, $rsrc, $args) = @_
+
+  ...your code...
+
+  __PACKAGE__->meta->make_immutable;
+
+  1;
+
+The L<MooseX::NonMoose> is necessary so that the L<Moose> constructor does not
+entirely overwrite the DBIC one (in contrast L<Moo> does this automatically).
+Alternatively, you can skip L<MooseX::NonMoose> and get by with just L<Moose>
+instead by doing:
+
+  __PACKAGE__->meta->make_immutable(inline_constructor => 0);
+
 =head1 METHODS
 
 =head2 new
@@ -232,12 +301,18 @@ creation B<will not work>. See also warning pertaining to L</create>.
 
 sub new {
   my $class = shift;
-  return $class->new_result(@_) if ref $class;
+
+  if (ref $class) {
+    DBIx::Class::_ENV_::ASSERT_NO_INTERNAL_INDIRECT_CALLS and fail_on_internal_call;
+    return $class->new_result(@_);
+  }
 
   my ($source, $attrs) = @_;
   $source = $source->resolve
     if $source->isa('DBIx::Class::ResultSourceHandle');
+
   $attrs = { %{$attrs||{}} };
+  delete @{$attrs}{qw(_last_sqlmaker_alias_map _simple_passthrough_construction)};
 
   if ($attrs->{page}) {
     $attrs->{rows} ||= 10;
@@ -295,8 +370,8 @@ 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>
+L<Searching|DBIx::Class::Manual::Cookbook/SEARCHING>. For a complete
+documentation for the first argument, see L<SQL::Abstract/"WHERE CLAUSES">
 and its extension L<DBIx::Class::SQLMaker>.
 
 For more help on using joins with search, see L<DBIx::Class::Manual::Joining>.
@@ -318,6 +393,7 @@ sub search {
   my $rs = $self->search_rs( @_ );
 
   if (wantarray) {
+    DBIx::Class::_ENV_::ASSERT_NO_INTERNAL_WANTARRAY and my $sog = fail_on_internal_wantarray;
     return $rs->all;
   }
   elsif (defined wantarray) {
@@ -390,18 +466,17 @@ sub search_rs {
   my $cache;
   my %safe = (alias => 1, cache => 1);
   if ( ! List::Util::first { !$safe{$_} } keys %$call_attrs and (
-    ! defined $_[0]
+    ! defined $call_cond
       or
-    ref $_[0] eq 'HASH' && ! keys %{$_[0]}
+    ref $call_cond eq 'HASH' && ! keys %$call_cond
       or
-    ref $_[0] eq 'ARRAY' && ! @{$_[0]}
+    ref $call_cond eq 'ARRAY' && ! @$call_cond
   )) {
     $cache = $self->get_cache;
   }
 
   my $old_attrs = { %{$self->{attrs}} };
-  my $old_having = delete $old_attrs->{having};
-  my $old_where = delete $old_attrs->{where};
+  my ($old_having, $old_where) = delete @{$old_attrs}{qw(having where)};
 
   my $new_attrs = { %$old_attrs };
 
@@ -436,6 +511,7 @@ sub search_rs {
 
     # older deprecated name, use only if {columns} is not there
     if (my $c = delete $new_attrs->{cols}) {
+      carp_unique( "Resultset attribute 'cols' is deprecated, use 'columns' instead" );
       if ($new_attrs->{columns}) {
         carp "Resultset specifies both the 'columns' and the legacy 'cols' attributes - ignoring 'cols'";
       }
@@ -482,8 +558,12 @@ sub _normalize_selection {
   my ($self, $attrs) = @_;
 
   # legacy syntax
-  $attrs->{'+columns'} = $self->_merge_attr($attrs->{'+columns'}, delete $attrs->{include_columns})
-    if exists $attrs->{include_columns};
+  if ( exists $attrs->{include_columns} ) {
+    carp_unique( "Resultset attribute 'include_columns' is deprecated, use '+columns' instead" );
+    $attrs->{'+columns'} = $self->_merge_attr(
+      $attrs->{'+columns'}, delete $attrs->{include_columns}
+    );
+  }
 
   # columns are always placed first, however
 
@@ -570,60 +650,32 @@ sub _normalize_selection {
 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 %$_;
+  (
+    (ref $_ eq 'ARRAY' and !@$_)
+      or
+    (ref $_ eq 'HASH' and ! keys %$_)
+  ) and $_ = undef for ($left, $right);
 
-      # 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'};
-        }
-      }
-    }
+  # either on of the two undef or both undef
+  if ( ( (defined $left) xor (defined $right) ) or ! defined $left ) {
+    return defined $left ? $left : $right;
   }
 
-  # merge hashes with weeding out of duplicates (simple cases only)
-  if (ref $left eq 'HASH' and ref $right eq 'HASH') {
+  my $cond = $self->result_source->schema->storage->_collapse_cond({ -and => [$left, $right] });
 
-    # 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->{$_} );
-    }
+  for my $c (grep { ref $cond->{$_} eq 'ARRAY' and ($cond->{$_}[0]||'') eq '-and' } keys %$cond) {
 
-    $right = undef unless keys %$right;
-  }
+    my @vals = sort @{$cond->{$c}}[ 1..$#{$cond->{$c}} ];
+    my @fin = shift @vals;
 
+    for my $v (@vals) {
+      push @fin, $v unless Data::Compare::Compare( $fin[-1], $v );
+    }
 
-  if (defined $left xor defined $right) {
-    return defined $left ? $left : $right;
-  }
-  elsif (! defined $left) {
-    return undef;
-  }
-  else {
-    return { -and => [ $left, $right ] };
+    $cond->{$c} = (@fin == 1) ? $fin[0] : [-and => @fin ];
   }
+
+  $cond;
 }
 
 =head2 search_literal
@@ -633,7 +685,7 @@ should only be used in that context. C<search_literal> is a convenience
 method. It is equivalent to calling C<< $schema->search(\[]) >>, but if you
 want to ensure columns are bound correctly, use L</search>.
 
-See L<DBIx::Class::Manual::Cookbook/Searching> and
+See L<DBIx::Class::Manual::Cookbook/SEARCHING> and
 L<DBIx::Class::Manual::FAQ/Searching> for searching techniques that do not
 require C<search_literal>.
 
@@ -784,11 +836,15 @@ sub find {
 
       next if $keyref eq 'ARRAY'; # has_many for multi_create
 
-      my $rel_q = $rsrc->_resolve_condition(
+      my ($rel_cond, $crosstable) = $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;
+
+      $self->throw_exception("Complex condition via relationship '$key' is unsupported in find()")
+         if $crosstable or ref($rel_cond) ne 'HASH';
+
+      # supplement
+      @related{keys %$rel_cond} = values %$rel_cond;
     }
   }
 
@@ -839,7 +895,7 @@ sub find {
 
   # Run the query, passing the result_class since it should propagate for find
   my $rs = $self->search ($final_cond, {result_class => $self->result_class, %$attrs});
-  if (keys %{$rs->_resolved_attrs->{collapse}}) {
+  if ($rs->_resolved_attrs->{collapse}) {
     my $row = $rs->next;
     carp "Query returned more than one row" if $rs->next;
     return $row;
@@ -991,7 +1047,7 @@ sub cursor {
   my $self = shift;
 
   return $self->{cursor} ||= do {
-    my $attrs = { %{$self->_resolved_attrs } };
+    my $attrs = $self->_resolved_attrs;
     $self->result_source->storage->select(
       $attrs->{from}, $attrs->{select}, $attrs->{where}, $attrs
     );
@@ -1049,11 +1105,9 @@ sub single {
 
   my $attrs = { %{$self->_resolved_attrs} };
 
-  if (keys %{$attrs->{collapse}}) {
-    $self->throw_exception(
-      'single() can not be used on resultsets prefetching has_many. Use find( \%cond ) or next() instead'
-    );
-  }
+  $self->throw_exception(
+    'single() can not be used on resultsets collapsing a has_many. Use find( \%cond ) or next() instead'
+  ) if $attrs->{collapse};
 
   if ($where) {
     if (defined $attrs->{where}) {
@@ -1067,45 +1121,14 @@ sub single {
     }
   }
 
-  my @data = $self->result_source->storage->select_single(
+  my $data = [ $self->result_source->storage->select_single(
     $attrs->{from}, $attrs->{select},
     $attrs->{where}, $attrs
-  );
-
-  return (@data ? ($self->_construct_object(@data))[0] : undef);
-}
-
-
-# _collapse_query
-#
-# Recursively collapse the query, accumulating values for each column.
-
-sub _collapse_query {
-  my ($self, $query, $collapsed) = @_;
-
-  $collapsed ||= {};
-
-  if (ref $query eq 'ARRAY') {
-    foreach my $subquery (@$query) {
-      next unless ref $subquery;  # -or
-      $collapsed = $self->_collapse_query($subquery, $collapsed);
-    }
-  }
-  elsif (ref $query eq 'HASH') {
-    if (keys %$query and (keys %$query)[0] eq '-and') {
-      foreach my $subquery (@{$query->{-and}}) {
-        $collapsed = $self->_collapse_query($subquery, $collapsed);
-      }
-    }
-    else {
-      foreach my $col (keys %$query) {
-        my $value = $query->{$col};
-        $collapsed->{$col}{$value}++;
-      }
-    }
-  }
+  )];
 
-  return $collapsed;
+  return undef unless @$data;
+  $self->{_stashed_rows} = [ $data ];
+  $self->_construct_results->[0];
 }
 
 =head2 get_column
@@ -1198,8 +1221,6 @@ sub slice {
   $attrs->{offset} += $min;
   $attrs->{rows} = ($max ? ($max - $min + 1) : 1);
   return $self->search(undef, $attrs);
-  #my $slice = (ref $self)->new($self->result_source, $attrs);
-  #return (wantarray ? $slice->all : $slice);
 }
 
 =head2 next
@@ -1229,161 +1250,292 @@ first record from the resultset.
 
 sub next {
   my ($self) = @_;
+
   if (my $cache = $self->get_cache) {
     $self->{all_cache_position} ||= 0;
     return $cache->[$self->{all_cache_position}++];
   }
+
   if ($self->{attrs}{cache}) {
     delete $self->{pager};
     $self->{all_cache_position} = 1;
     return ($self->all)[0];
   }
-  if ($self->{stashed_objects}) {
-    my $obj = shift(@{$self->{stashed_objects}});
-    delete $self->{stashed_objects} unless @{$self->{stashed_objects}};
-    return $obj;
-  }
-  my @row = (
-    exists $self->{stashed_row}
-      ? @{delete $self->{stashed_row}}
-      : $self->cursor->next
-  );
-  return undef unless (@row);
-  my ($row, @more) = $self->_construct_object(@row);
-  $self->{stashed_objects} = \@more if @more;
-  return $row;
-}
 
-sub _construct_object {
-  my ($self, @row) = @_;
+  return shift(@{$self->{_stashed_results}}) if @{ $self->{_stashed_results}||[] };
 
-  my $info = $self->_collapse_result($self->{_attrs}{as}, \@row)
-    or return ();
-  my @new = $self->result_class->inflate_result($self->result_source, @$info);
-  @new = $self->{_attrs}{record_filter}->(@new)
-    if exists $self->{_attrs}{record_filter};
-  return @new;
-}
+  $self->{_stashed_results} = $self->_construct_results
+    or return undef;
 
-sub _collapse_result {
-  my ($self, $as_proto, $row) = @_;
+  return shift @{$self->{_stashed_results}};
+}
 
-  my @copy = @$row;
+# Constructs as many results as it can in one pass while respecting
+# cursor laziness. Several modes of operation:
+#
+# * Always builds everything present in @{$self->{_stashed_rows}}
+# * If called with $fetch_all true - pulls everything off the cursor and
+#   builds all result structures (or objects) in one pass
+# * If $self->_resolved_attrs->{collapse} is true, checks the order_by
+#   and if the resultset is ordered properly by the left side:
+#   * Fetches stuff off the cursor until the "master object" changes,
+#     and saves the last extra row (if any) in @{$self->{_stashed_rows}}
+#   OR
+#   * Just fetches, and collapses/constructs everything as if $fetch_all
+#     was requested (there is no other way to collapse except for an
+#     eager cursor)
+# * If no collapse is requested - just get the next row, construct and
+#   return
+sub _construct_results {
+  my ($self, $fetch_all) = @_;
 
-  # 'foo'         => [ undef, 'foo' ]
-  # 'foo.bar'     => [ 'foo', 'bar' ]
-  # 'foo.bar.baz' => [ 'foo.bar', 'baz' ]
+  my $rsrc = $self->result_source;
+  my $attrs = $self->_resolved_attrs;
 
-  my @construct_as = map { [ (/^(?:(.*)\.)?([^.]+)$/) ] } @$as_proto;
+  if (
+    ! $fetch_all
+      and
+    ! $attrs->{order_by}
+      and
+    $attrs->{collapse}
+      and
+    my @pcols = $rsrc->primary_columns
+  ) {
+    # default order for collapsing unless the user asked for something
+    $attrs->{order_by} = [ map { join '.', $attrs->{alias}, $_} @pcols ];
+    $attrs->{_ordered_for_collapse} = 1;
+    $attrs->{_order_is_artificial} = 1;
+  }
 
-  my %collapse = %{$self->{_attrs}{collapse}||{}};
+  # this will be used as both initial raw-row collector AND as a RV of
+  # _construct_results. Not regrowing the array twice matters a lot...
+  # a surprising amount actually
+  my $rows = delete $self->{_stashed_rows};
 
-  my @pri_index;
+  my $cursor; # we may not need one at all
 
-  # if we're doing collapsing (has_many prefetch) we need to grab records
-  # until the PK changes, so fill @pri_index. if not, we leave it empty so
-  # we know we don't have to bother.
+  my $did_fetch_all = $fetch_all;
 
-  # the reason for not using the collapse stuff directly is because if you
-  # had for e.g. two artists in a row with no cds, the collapse info for
-  # both would be NULL (undef) so you'd lose the second artist
+  if ($fetch_all) {
+    # FIXME SUBOPTIMAL - we can do better, cursor->next/all (well diff. methods) should return a ref
+    $rows = [ ($rows ? @$rows : ()), $self->cursor->all ];
+  }
+  elsif( $attrs->{collapse} ) {
 
-  # store just the index so we can check the array positions from the row
-  # without having to contruct the full hash
+    # a cursor will need to be closed over in case of collapse
+    $cursor = $self->cursor;
 
-  if (keys %collapse) {
-    my %pri = map { ($_ => 1) } $self->result_source->_pri_cols;
-    foreach my $i (0 .. $#construct_as) {
-      next if defined($construct_as[$i][0]); # only self table
-      if (delete $pri{$construct_as[$i][1]}) {
-        push(@pri_index, $i);
+    $attrs->{_ordered_for_collapse} = (
+      (
+        $attrs->{order_by}
+          and
+        $rsrc->schema
+              ->storage
+               ->_extract_colinfo_of_stable_main_source_order_by_portion($attrs)
+      ) ? 1 : 0
+    ) unless defined $attrs->{_ordered_for_collapse};
+
+    if (! $attrs->{_ordered_for_collapse}) {
+      $did_fetch_all = 1;
+
+      # instead of looping over ->next, use ->all in stealth mode
+      # *without* calling a ->reset afterwards
+      # FIXME ENCAPSULATION - encapsulation breach, cursor method additions pending
+      if (! $cursor->{_done}) {
+        $rows = [ ($rows ? @$rows : ()), $cursor->all ];
+        $cursor->{_done} = 1;
       }
-      last unless keys %pri; # short circuit (Johnny Five Is Alive!)
     }
   }
 
-  # no need to do an if, it'll be empty if @pri_index is empty anyway
+  if (! $did_fetch_all and ! @{$rows||[]} ) {
+    # FIXME SUBOPTIMAL - we can do better, cursor->next/all (well diff. methods) should return a ref
+    $cursor ||= $self->cursor;
+    if (scalar (my @r = $cursor->next) ) {
+      $rows = [ \@r ];
+    }
+  }
+
+  return undef unless @{$rows||[]};
 
-  my %pri_vals = map { ($_ => $copy[$_]) } @pri_index;
+  # sanity check - people are too clever for their own good
+  if ($attrs->{collapse} and my $aliastypes = $attrs->{_last_sqlmaker_alias_map} ) {
 
-  my @const_rows;
+    my $multiplied_selectors;
+    for my $sel_alias ( grep { $_ ne $attrs->{alias} } keys %{ $aliastypes->{selecting} } ) {
+      if (
+        $aliastypes->{multiplying}{$sel_alias}
+          or
+        $aliastypes->{premultiplied}{$sel_alias}
+      ) {
+        $multiplied_selectors->{$_} = 1 for values %{$aliastypes->{selecting}{$sel_alias}{-seen_columns}}
+      }
+    }
 
-  do { # no need to check anything at the front, we always want the first row
+    for my $i (0 .. $#{$attrs->{as}} ) {
+      my $sel = $attrs->{select}[$i];
 
-    my %const;
+      if (ref $sel eq 'SCALAR') {
+        $sel = $$sel;
+      }
+      elsif( ref $sel eq 'REF' and ref $$sel eq 'ARRAY' ) {
+        $sel = $$sel->[0];
+      }
 
-    foreach my $this_as (@construct_as) {
-      $const{$this_as->[0]||''}{$this_as->[1]} = shift(@copy);
+      $self->throw_exception(
+        'Result collapse not possible - selection from a has_many source redirected to the main object'
+      ) if ($multiplied_selectors->{$sel} and $attrs->{as}[$i] !~ /\./);
     }
+  }
 
-    push(@const_rows, \%const);
+  # hotspot - skip the setter
+  my $res_class = $self->_result_class;
 
-  } until ( # no pri_index => no collapse => drop straight out
-      !@pri_index
-    or
-      do { # get another row, stash it, drop out if different PK
+  my $inflator_cref = $self->{_result_inflator}{cref} ||= do {
+    $res_class->can ('inflate_result')
+      or $self->throw_exception("Inflator $res_class does not provide an inflate_result() method");
+  };
+
+  my $infmap = $attrs->{as};
 
-        @copy = $self->cursor->next;
-        $self->{stashed_row} = \@copy;
+  $self->{_result_inflator}{is_core_row} = ( (
+    $inflator_cref
+      ==
+    ( \&DBIx::Class::Row::inflate_result || die "No ::Row::inflate_result() - can't happen" )
+  ) ? 1 : 0 ) unless defined $self->{_result_inflator}{is_core_row};
 
-        # last thing in do block, counts as true if anything doesn't match
+  $self->{_result_inflator}{is_hri} = ( (
+    ! $self->{_result_inflator}{is_core_row}
+      and
+    $inflator_cref == (
+      require DBIx::Class::ResultClass::HashRefInflator
+        &&
+      DBIx::Class::ResultClass::HashRefInflator->can('inflate_result')
+    )
+  ) ? 1 : 0 ) unless defined $self->{_result_inflator}{is_hri};
 
-        # check xor defined first for NULL vs. NOT NULL then if one is
-        # defined the other must be so check string equality
 
-        grep {
-          (defined $pri_vals{$_} ^ defined $copy[$_])
-          || (defined $pri_vals{$_} && ($pri_vals{$_} ne $copy[$_]))
-        } @pri_index;
+  if ($attrs->{_simple_passthrough_construction}) {
+    # construct a much simpler array->hash folder for the one-table HRI cases right here
+    if ($self->{_result_inflator}{is_hri}) {
+      for my $r (@$rows) {
+        $r = { map { $infmap->[$_] => $r->[$_] } 0..$#$infmap };
       }
-  );
+    }
+    # FIXME SUBOPTIMAL this is a very very very hot spot
+    # while rather optimal we can *still* do much better, by
+    # building a smarter Row::inflate_result(), and
+    # switch to feeding it data via a much leaner interface
+    #
+    # crude unscientific benchmarking indicated the shortcut eval is not worth it for
+    # this particular resultset size
+    elsif ( $self->{_result_inflator}{is_core_row} and @$rows < 60 ) {
+      for my $r (@$rows) {
+        $r = $inflator_cref->($res_class, $rsrc, { map { $infmap->[$_] => $r->[$_] } (0..$#$infmap) } );
+      }
+    }
+    else {
+      eval sprintf (
+        ( $self->{_result_inflator}{is_core_row}
+          ? '$_ = $inflator_cref->($res_class, $rsrc, { %s }) for @$rows'
+          # a custom inflator may be a multiplier/reductor - put it in direct list ctx
+          : '@$rows = map { $inflator_cref->($res_class, $rsrc, { %s } ) } @$rows'
+        ),
+        ( join (', ', map { "\$infmap->[$_] => \$_->[$_]" } 0..$#$infmap ) )
+      );
+    }
+  }
+  else {
+    my $parser_type =
+        $self->{_result_inflator}{is_hri}       ? 'hri'
+      : $self->{_result_inflator}{is_core_row}  ? 'classic_pruning'
+      :                                           'classic_nonpruning'
+    ;
 
-  my $alias = $self->{attrs}{alias};
-  my $info = [];
+    # $args and $attrs to _mk_row_parser are separated to delineate what is
+    # core collapser stuff and what is dbic $rs specific
+    @{$self->{_row_parser}{$parser_type}}{qw(cref nullcheck)} = $rsrc->_mk_row_parser({
+      eval => 1,
+      inflate_map => $infmap,
+      collapse => $attrs->{collapse},
+      premultiplied => $attrs->{_main_source_premultiplied},
+      hri_style => $self->{_result_inflator}{is_hri},
+      prune_null_branches => $self->{_result_inflator}{is_hri} || $self->{_result_inflator}{is_core_row},
+    }, $attrs) unless $self->{_row_parser}{$parser_type}{cref};
+
+    # column_info metadata historically hasn't been too reliable.
+    # We need to start fixing this somehow (the collapse resolver
+    # can't work without it). Add an explicit check for the *main*
+    # result, hopefully this will gradually weed out such errors
+    #
+    # FIXME - this is a temporary kludge that reduces performance
+    # It is however necessary for the time being
+    my ($unrolled_non_null_cols_to_check, $err);
+
+    if (my $check_non_null_cols = $self->{_row_parser}{$parser_type}{nullcheck} ) {
+
+      $err =
+        'Collapse aborted due to invalid ResultSource metadata - the following '
+      . 'selections are declared non-nullable but NULLs were retrieved: '
+      ;
 
-  my %collapse_pos;
+      my @violating_idx;
+      COL: for my $i (@$check_non_null_cols) {
+        ! defined $_->[$i] and push @violating_idx, $i and next COL for @$rows;
+      }
 
-  my @const_keys;
+      $self->throw_exception( $err . join (', ', map { "'$infmap->[$_]'" } @violating_idx ) )
+        if @violating_idx;
 
-  foreach my $const (@const_rows) {
-    scalar @const_keys or do {
-      @const_keys = sort { length($a) <=> length($b) } keys %$const;
-    };
-    foreach my $key (@const_keys) {
-      if (length $key) {
-        my $target = $info;
-        my @parts = split(/\./, $key);
-        my $cur = '';
-        my $data = $const->{$key};
-        foreach my $p (@parts) {
-          $target = $target->[1]->{$p} ||= [];
-          $cur .= ".${p}";
-          if ($cur eq ".${key}" && (my @ckey = @{$collapse{$cur}||[]})) {
-            # collapsing at this point and on final part
-            my $pos = $collapse_pos{$cur};
-            CK: foreach my $ck (@ckey) {
-              if (!defined $pos->{$ck} || $pos->{$ck} ne $data->{$ck}) {
-                $collapse_pos{$cur} = $data;
-                delete @collapse_pos{ # clear all positioning for sub-entries
-                  grep { m/^\Q${cur}.\E/ } keys %collapse_pos
-                };
-                push(@$target, []);
-                last CK;
-              }
-            }
-          }
-          if (exists $collapse{$cur}) {
-            $target = $target->[-1];
-          }
-        }
-        $target->[0] = $data;
-      } else {
-        $info->[0] = $const->{$key};
+      $unrolled_non_null_cols_to_check = join (',', @$check_non_null_cols);
+    }
+
+    my $next_cref =
+      ($did_fetch_all or ! $attrs->{collapse})  ? undef
+    : defined $unrolled_non_null_cols_to_check  ? eval sprintf <<'EOS', $unrolled_non_null_cols_to_check
+sub {
+  # FIXME SUBOPTIMAL - we can do better, cursor->next/all (well diff. methods) should return a ref
+  my @r = $cursor->next or return;
+  if (my @violating_idx = grep { ! defined $r[$_] } (%s) ) {
+    $self->throw_exception( $err . join (', ', map { "'$infmap->[$_]'" } @violating_idx ) )
+  }
+  \@r
+}
+EOS
+    : sub {
+        # FIXME SUBOPTIMAL - we can do better, cursor->next/all (well diff. methods) should return a ref
+        my @r = $cursor->next or return;
+        \@r
       }
+    ;
+
+    $self->{_row_parser}{$parser_type}{cref}->(
+      $rows,
+      $next_cref ? ( $next_cref, $self->{_stashed_rows} = [] ) : (),
+    );
+
+    # simple in-place substitution, does not regrow $rows
+    if ($self->{_result_inflator}{is_core_row}) {
+      $_ = $inflator_cref->($res_class, $rsrc, @$_) for @$rows
+    }
+    # Special-case multi-object HRI - there is no $inflator_cref pass at all
+    elsif ( ! $self->{_result_inflator}{is_hri} ) {
+      # the inflator may be a multiplier/reductor - put it in list ctx
+      @$rows = map { $inflator_cref->($res_class, $rsrc, @$_) } @$rows;
     }
   }
 
-  return $info;
+  # The @$rows check seems odd at first - why wouldn't we want to warn
+  # regardless? The issue is things like find() etc, where the user
+  # *knows* only one result will come back. In these cases the ->all
+  # is not a pessimization, but rather something we actually want
+  carp_unique(
+    'Unable to properly collapse has_many results in iterator mode due '
+  . 'to order criteria - performed an eager cursor slurp underneath. '
+  . 'Consider using ->all() instead'
+  ) if ( ! $fetch_all and @$rows > 1 );
+
+  return $rows;
 }
 
 =head2 result_source
@@ -1423,14 +1575,22 @@ in the original source class will not run.
 sub result_class {
   my ($self, $result_class) = @_;
   if ($result_class) {
-    unless (ref $result_class) { # don't fire this for an object
-      $self->ensure_class_loaded($result_class);
+
+    # don't fire this for an object
+    $self->ensure_class_loaded($result_class)
+      unless ref($result_class);
+
+    if ($self->get_cache) {
+      carp_unique('Changing the result_class of a ResultSet instance with cached results is a noop - the cache contents will not be altered');
+    }
+    # FIXME ENCAPSULATION - encapsulation breach, cursor method additions pending
+    elsif ($self->{cursor} && $self->{cursor}{_pos}) {
+      $self->throw_exception('Changing the result_class of a ResultSet instance with an active cursor is not supported');
     }
+
     $self->_result_class($result_class);
-    # THIS LINE WOULD BE A BUG - this accessor specifically exists to
-    # permit the user to set result class on one result set only; it only
-    # chains if provided to search()
-    #$self->{attrs}{result_class} = $result_class if ref $self;
+
+    delete $self->{_result_inflator};
   }
   $self->_result_class;
 }
@@ -1460,8 +1620,7 @@ sub count {
 
   # this is a little optimization - it is faster to do the limit
   # adjustments in software, instead of a subquery
-  my $rows = delete $attrs->{rows};
-  my $offset = delete $attrs->{offset};
+  my ($rows, $offset) = delete @{$attrs}{qw/rows offset/};
 
   my $crs;
   if ($self->_has_resolved_attr (qw/collapse group_by/)) {
@@ -1509,10 +1668,10 @@ sub count_rs {
   # software based limiting can not be ported if this $rs is to be used
   # in a subquery itself (i.e. ->as_query)
   if ($self->_has_resolved_attr (qw/collapse group_by offset rows/)) {
-    return $self->_count_subq_rs;
+    return $self->_count_subq_rs($self->{_attrs});
   }
   else {
-    return $self->_count_rs;
+    return $self->_count_rs($self->{_attrs});
   }
 }
 
@@ -1523,20 +1682,17 @@ sub _count_rs {
   my ($self, $attrs) = @_;
 
   my $rsrc = $self->result_source;
-  $attrs ||= $self->_resolved_attrs;
 
   my $tmp_attrs = { %$attrs };
   # take off any limits, record_filter is cdbi, and no point of ordering nor locking a count
   delete @{$tmp_attrs}{qw/rows offset order_by record_filter for/};
 
   # overwrite the selector (supplied by the storage)
-  $tmp_attrs->{select} = $rsrc->storage->_count_select ($rsrc, $attrs);
-  $tmp_attrs->{as} = 'count';
-  delete @{$tmp_attrs}{qw/columns/};
-
-  my $tmp_rs = $rsrc->resultset_class->new($rsrc, $tmp_attrs)->get_column ('count');
-
-  return $tmp_rs;
+  $rsrc->resultset_class->new($rsrc, {
+    %$tmp_attrs,
+    select => $rsrc->storage->_count_select ($rsrc, $attrs),
+    as => 'count',
+  })->get_column ('count');
 }
 
 #
@@ -1546,15 +1702,14 @@ sub _count_subq_rs {
   my ($self, $attrs) = @_;
 
   my $rsrc = $self->result_source;
-  $attrs ||= $self->_resolved_attrs;
 
   my $sub_attrs = { %$attrs };
   # extra selectors do not go in the subquery and there is no point of ordering it, nor locking it
-  delete @{$sub_attrs}{qw/collapse columns as select _prefetch_selector_range order_by for/};
+  delete @{$sub_attrs}{qw/collapse columns as select order_by for/};
 
   # 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 ( keys %{$attrs->{collapse}}  ) {
+  if ( $attrs->{collapse}  ) {
     $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 '
@@ -1592,18 +1747,22 @@ sub _count_subq_rs {
 
       my ($lquote, $rquote, $sep) = map { quotemeta $_ } ($sql_maker->_quote_chars, $sql_maker->name_sep);
 
-      my $sql = $sql_maker->_parse_rs_attrs ({ having => $attrs->{having} });
+      my $having_sql = $sql_maker->_parse_rs_attrs ({ having => $attrs->{having} });
+      my %seen_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 =~ /
+      while ($having_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
+        my $part = $1 || $2 || $3;  # one of them matched if we got here
+        unless ($seen_having{$part}++) {
+          push @parts, $part;
+        }
       }
     }
 
@@ -1633,9 +1792,6 @@ sub _count_subq_rs {
                   ->get_column ('count');
 }
 
-sub _bool {
-  return 1;
-}
 
 =head2 count_literal
 
@@ -1674,33 +1830,22 @@ Returns all elements in the resultset.
 sub all {
   my $self = shift;
   if(@_) {
-      $self->throw_exception("all() doesn't take any arguments, you probably wanted ->search(...)->all()");
+    $self->throw_exception("all() doesn't take any arguments, you probably wanted ->search(...)->all()");
   }
 
-  return @{ $self->get_cache } if $self->get_cache;
-
-  my @obj;
-
-  if (keys %{$self->_resolved_attrs->{collapse}}) {
-    # Using $self->cursor->all is really just an optimisation.
-    # If we're collapsing has_many prefetches it probably makes
-    # very little difference, and this is cleaner than hacking
-    # _construct_object to survive the approach
-    $self->cursor->reset;
-    my @row = $self->cursor->next;
-    while (@row) {
-      push(@obj, $self->_construct_object(@row));
-      @row = (exists $self->{stashed_row}
-               ? @{delete $self->{stashed_row}}
-               : $self->cursor->next);
-    }
-  } else {
-    @obj = map { $self->_construct_object(@$_) } $self->cursor->all;
+  delete @{$self}{qw/_stashed_rows _stashed_results/};
+
+  if (my $c = $self->get_cache) {
+    return @$c;
   }
 
-  $self->set_cache(\@obj) if $self->{attrs}{cache};
+  $self->cursor->reset;
+
+  my $objs = $self->_construct_results('fetch_all') || [];
 
-  return @obj;
+  $self->set_cache($objs) if $self->{attrs}{cache};
+
+  return @$objs;
 }
 
 =head2 reset
@@ -1721,6 +1866,8 @@ another query.
 
 sub reset {
   my ($self) = @_;
+
+  delete @{$self}{qw/_stashed_rows _stashed_results/};
   $self->{all_cache_position} = 0;
   $self->cursor->reset;
   return $self;
@@ -1755,14 +1902,13 @@ sub first {
 sub _rs_update_delete {
   my ($self, $op, $values) = @_;
 
-  my $cond = $self->{cond};
   my $rsrc = $self->result_source;
   my $storage = $rsrc->schema->storage;
 
   my $attrs = { %{$self->_resolved_attrs} };
 
   my $join_classifications;
-  my $existing_group_by = delete $attrs->{group_by};
+  my ($existing_group_by) = delete @{$attrs}{qw(group_by _grouped_by_distinct)};
 
   # do we need a subquery for any reason?
   my $needs_subq = (
@@ -1777,20 +1923,12 @@ sub _rs_update_delete {
 
   # simplify the joinmap, so we can further decide if a subq is necessary
   if (!$needs_subq and @{$attrs->{from}} > 1) {
-    $attrs->{from} = $storage->_prune_unused_joins ($attrs->{from}, $attrs->{select}, $cond, $attrs);
-
-    # check if there are any joins left after the prune
-    if ( @{$attrs->{from}} > 1 ) {
-      $join_classifications = $storage->_resolve_aliastypes_from_select_args (
-        [ @{$attrs->{from}}[1 .. $#{$attrs->{from}}] ],
-        $attrs->{select},
-        $cond,
-        $attrs
-      );
 
-      # any non-pruneable joins imply subq
-      $needs_subq = scalar keys %{ $join_classifications->{restricting} || {} };
-    }
+    ($attrs->{from}, $join_classifications) =
+      $storage->_prune_unused_joins ($attrs);
+
+    # any non-pruneable non-local restricting joins imply subq
+    $needs_subq = defined List::Util::first { $_ ne $attrs->{alias} } keys %{ $join_classifications->{restricting} || {} };
   }
 
   # check if the head is composite (by now all joins are thrown out unless $needs_subq)
@@ -1800,115 +1938,106 @@ sub _rs_update_delete {
     ref $attrs->{from}[0]{ $attrs->{from}[0]{-alias} }
   );
 
+  my ($cond, $guard);
   # do we need anything like a subquery?
-  unless ($needs_subq) {
+  if (! $needs_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. Tell SQLMaker to dequalify idents via a gross hack.
-    my $cond = do {
+    $cond = do {
       my $sqla = $rsrc->storage->sql_maker;
       local $sqla->{_dequalify_idents} = 1;
       \[ $sqla->_recurse_where($self->{cond}) ];
     };
-
-    return $rsrc->storage->$op(
-      $rsrc,
-      $op eq 'update' ? $values : (),
-      $cond,
-    );
   }
+  else {
+    # 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,
+      )
+    );
 
-  # 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,
-    )
-  );
+    # make a new $rs selecting only the PKs (that's all we really need for the subq)
+    delete $attrs->{$_} for qw/select as collapse/;
+    $attrs->{columns} = [ map { "$attrs->{alias}.$_" } @$idcols ];
 
-  # 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) {
-    return $storage->$op (
-      $rsrc,
-      $op eq 'update' ? $values : (),
+    # this will be consumed by the pruner waaaaay down the stack
+    $attrs->{_force_prune_multiplying_joins} = 1;
+
+    my $subrs = (ref $self)->new($rsrc, $attrs);
+
+    if (@$idcols == 1) {
+      $cond = { $idcols->[0] => { -in => $subrs->as_query } };
+    }
+    elsif ($storage->_use_multicolumn_in) {
       # no syntax for calling this properly yet
       # !!! EXPERIMENTAL API !!! WILL CHANGE !!!
-      $storage->sql_maker->_where_op_multicolumn_in (
+      $cond = $storage->sql_maker->_where_op_multicolumn_in (
         $idcols, # how do I convey a list of idents...? can binds reside on lhs?
         $subrs->as_query
       ),
-    );
-  }
-  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/multiplication starts to matter
-    if (
-      $existing_group_by
-        or
-      keys %{ $join_classifications->{multiplying} || {} }
-    ) {
-      # 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}.$_" }
-          @$existing_group_by
-        ;
-
-        if (
-          join ("\x00", sort @current_group_by)
-            ne
-          join ("\x00", sort @{$attrs->{columns}} )
-        ) {
-          $self->throw_exception (
-            "You have just attempted a $op operation on a resultset which does group_by"
-            . ' on columns other than the primary keys, while DBIC internally needs to retrieve'
-            . ' the primary keys in a subselect. All sane RDBMS engines do not support this'
-            . ' kind of queries. Please retry the operation with a modified group_by or'
-            . ' without using one at all.'
-          );
+    }
+    else {
+      # 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/multiplication starts to matter
+      if (
+        $existing_group_by
+          or
+        # we do not need to check pre-multipliers, since if the premulti is there, its
+        # parent (who is multi) will be there too
+        keys %{ $join_classifications->{multiplying} || {} }
+      ) {
+        # 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}.$_" }
+            @$existing_group_by
+          ;
+
+          if (
+            join ("\x00", sort @current_group_by)
+              ne
+            join ("\x00", sort @{$attrs->{columns}} )
+          ) {
+            $self->throw_exception (
+              "You have just attempted a $op operation on a resultset which does group_by"
+              . ' on columns other than the primary keys, while DBIC internally needs to retrieve'
+              . ' the primary keys in a subselect. All sane RDBMS engines do not support this'
+              . ' kind of queries. Please retry the operation with a modified group_by or'
+              . ' without using one at all.'
+            );
+          }
         }
-      }
 
-      $subrs = $subrs->search({}, { group_by => $attrs->{columns} });
-    }
+        $subrs = $subrs->search({}, { group_by => $attrs->{columns} });
+      }
 
-    my $guard = $storage->txn_scope_guard;
+      $guard = $storage->txn_scope_guard;
 
-    my @op_condition;
-    for my $row ($subrs->cursor->all) {
-      push @op_condition, { map
-        { $idcols->[$_] => $row->[$_] }
-        (0 .. $#$idcols)
-      };
+      for my $row ($subrs->cursor->all) {
+        push @$cond, { map
+          { $idcols->[$_] => $row->[$_] }
+          (0 .. $#$idcols)
+        };
+      }
     }
+  }
 
-    my $res = $storage->$op (
-      $rsrc,
-      $op eq 'update' ? $values : (),
-      \@op_condition,
-    );
+  my $res = $cond ? $storage->$op (
+    $rsrc,
+    $op eq 'update' ? $values : (),
+    $cond,
+  ) : '0E0';
 
-    $guard->commit;
+  $guard->commit if $guard;
 
-    return $res;
-  }
+  return $res;
 }
 
 =head2 update
@@ -2078,7 +2207,7 @@ first element should be a list of column names and each subsequent
 element should be a data value in the earlier specified column order.
 For example:
 
-  $Arstist_rs->populate([
+  $schema->resultset("Artist")->populate([
     [ qw( artistid name ) ],
     [ 100, 'A Formally Unknown Singer' ],
     [ 101, 'A singer that jumped the shark two albums ago' ],
@@ -2114,136 +2243,275 @@ case there are obviously no benefits to using this method over L</create>.
 sub populate {
   my $self = shift;
 
-  # cruft placed in standalone method
-  my $data = $self->_normalize_populate_args(@_);
+  my ($data, $guard);
 
-  return unless @$data;
+  # this is naive and just a quick check
+  # the types will need to be checked more thoroughly when the
+  # multi-source populate gets added
+  if (ref $_[0] eq 'ARRAY') {
+    return unless @{$_[0]};
 
-  if(defined wantarray) {
-    my @created = map { $self->create($_) } @$data;
-    return wantarray ? @created : \@created;
+    $data = $_[0] if (ref $_[0][0] eq 'HASH' or ref $_[0][0] eq 'ARRAY');
   }
-  else {
-    my $first = $data->[0];
 
-    # if a column is a registered relationship, and is a non-blessed hash/array, consider
-    # it relationship data
-    my (@rels, @columns);
-    my $rsrc = $self->result_source;
-    my $rels = { map { $_ => $rsrc->relationship_info($_) } $rsrc->relationships };
-    for (keys %$first) {
-      my $ref = ref $first->{$_};
-      $rels->{$_} && ($ref eq 'ARRAY' or $ref eq 'HASH')
-        ? push @rels, $_
-        : push @columns, $_
-      ;
-    }
+  $self->throw_exception('Populate expects an arrayref of hashrefs or arrayref of arrayrefs')
+    unless $data;
 
-    my @pks = $rsrc->primary_columns;
+  # FIXME - no cref handling
+  # At this point assume either hashes or arrays
 
-    ## do the belongs_to relationships
-    foreach my $index (0..$#$data) {
+  if(defined wantarray) {
+    my @results;
 
-      # delegate to create() for any dataset without primary keys with specified relationships
-      if (grep { !defined $data->[$index]->{$_} } @pks ) {
-        for my $r (@rels) {
-          if (grep { ref $data->[$index]{$r} eq $_ } qw/HASH ARRAY/) {  # a related set must be a HASH or AoH
-            my @ret = $self->populate($data);
-            return;
-          }
-        }
-      }
+    $guard = $self->result_source->schema->storage->txn_scope_guard
+      if ( @$data > 2 or ( @$data == 2 and ref $data->[0] eq 'ARRAY' ) );
 
-      foreach my $rel (@rels) {
-        next unless ref $data->[$index]->{$rel} eq "HASH";
-        my $result = $self->related_resultset($rel)->create($data->[$index]->{$rel});
-        my ($reverse_relname, $reverse_relinfo) = %{$rsrc->reverse_relationship_info($rel)};
-        my $related = $result->result_source->_resolve_condition(
-          $reverse_relinfo->{cond},
-          $self,
-          $result,
-          $rel,
-        );
-
-        delete $data->[$index]->{$rel};
-        $data->[$index] = {%{$data->[$index]}, %$related};
-
-        push @columns, keys %$related if $index == 0;
-      }
+    if (ref $data->[0] eq 'ARRAY') {
+      @results = map
+        { my $vals = $_; $self->new_result({ map { $data->[0][$_] => $vals->[$_] } 0..$#{$data->[0]} })->insert }
+        @{$data}[1 .. $#$data]
+      ;
+    }
+    else {
+      @results = map { $self->new_result($_)->insert } @$data;
     }
 
-    ## inherit the data locked in the conditions of the resultset
-    my ($rs_data) = $self->_merge_with_rscond({});
-    delete @{$rs_data}{@columns};
+    $guard->commit if $guard;
+    return wantarray ? @results : \@results;
+  }
 
-    ## do bulk insert on current row
-    $rsrc->storage->insert_bulk(
-      $rsrc,
-      [@columns, keys %$rs_data],
-      [ map { [ @$_{@columns}, values %$rs_data ] } @$data ],
-    );
+  # we have to deal with *possibly incomplete* related data
+  # this means we have to walk the data structure twice
+  # whether we want this or not
+  # jnap, I hate you ;)
+  my $rsrc = $self->result_source;
+  my $rel_info = { map { $_ => $rsrc->relationship_info($_) } $rsrc->relationships };
 
-    ## do the has_many relationships
-    foreach my $item (@$data) {
+  my ($colinfo, $colnames, $slices_with_rels);
+  my $data_start = 0;
 
-      my $main_row;
+  DATA_SLICE:
+  for my $i (0 .. $#$data) {
 
-      foreach my $rel (@rels) {
-        next unless ref $item->{$rel} eq "ARRAY" && @{ $item->{$rel} };
+    my $current_slice_seen_rel_infos;
 
-        $main_row ||= $self->new_result({map { $_ => $item->{$_} } @pks});
+### Determine/Supplement collists
+### BEWARE - This is a hot piece of code, a lot of weird idioms were used
+    if( ref $data->[$i] eq 'ARRAY' ) {
 
-        my $child = $main_row->$rel;
+      # positional(!) explicit column list
+      if ($i == 0) {
+
+        $colinfo->{$data->[0][$_]} = { pos => $_, name => $data->[0][$_] } and push @$colnames, $data->[0][$_]
+          for 0 .. $#{$data->[0]};
+
+        $data_start = 1;
+
+        next DATA_SLICE;
+      }
+      else {
+        for (values %$colinfo) {
+          if ($_->{is_rel} ||= (
+            $rel_info->{$_->{name}}
+              and
+            (
+              ref $data->[$i][$_->{pos}] eq 'ARRAY'
+                or
+              ref $data->[$i][$_->{pos}] eq 'HASH'
+                or
+              ( defined blessed $data->[$i][$_->{pos}] and $data->[$i][$_->{pos}]->isa('DBIx::Class::Row') )
+            )
+              and
+            1
+          )) {
+
+            # moar sanity check... sigh
+            for ( ref $data->[$i][$_->{pos}] eq 'ARRAY' ? @{$data->[$i][$_->{pos}]} : $data->[$i][$_->{pos}] ) {
+              if ( defined blessed $_ and $_->isa('DBIx::Class::Row' ) ) {
+                carp_unique("Fast-path populate() with supplied related objects is not possible - falling back to regular create()");
+                return my $throwaway = $self->populate(@_);
+              }
+            }
 
-        my $related = $child->result_source->_resolve_condition(
-          $rels->{$rel}{cond},
-          $child,
-          $main_row,
-          $rel,
-        );
+            push @$current_slice_seen_rel_infos, $rel_info->{$_->{name}};
+          }
+        }
+      }
 
-        my @rows_to_add = ref $item->{$rel} eq 'ARRAY' ? @{$item->{$rel}} : ($item->{$rel});
-        my @populate = map { {%$_, %$related} } @rows_to_add;
+     if ($current_slice_seen_rel_infos) {
+        push @$slices_with_rels, { map { $colnames->[$_] => $data->[$i][$_] } 0 .. $#$colnames };
 
-        $child->populate( \@populate );
+        # this is needed further down to decide whether or not to fallback to create()
+        $colinfo->{$colnames->[$_]}{seen_null} ||= ! defined $data->[$i][$_]
+          for 0 .. $#$colnames;
       }
     }
-  }
-}
+    elsif( ref $data->[$i] eq 'HASH' ) {
+
+      for ( sort keys %{$data->[$i]} ) {
+
+        $colinfo->{$_} ||= do {
+
+          $self->throw_exception("Column '$_' must be present in supplied explicit column list")
+            if $data_start; # it will be 0 on AoH, 1 on AoA
+
+          push @$colnames, $_;
+
+          # RV
+          { pos => $#$colnames, name => $_ }
+        };
+
+        if ($colinfo->{$_}{is_rel} ||= (
+          $rel_info->{$_}
+            and
+          (
+            ref $data->[$i]{$_} eq 'ARRAY'
+              or
+            ref $data->[$i]{$_} eq 'HASH'
+              or
+            ( defined blessed $data->[$i]{$_} and $data->[$i]{$_}->isa('DBIx::Class::Row') )
+          )
+            and
+          1
+        )) {
+
+          # moar sanity check... sigh
+          for ( ref $data->[$i]{$_} eq 'ARRAY' ? @{$data->[$i]{$_}} : $data->[$i]{$_} ) {
+            if ( defined blessed $_ and $_->isa('DBIx::Class::Row' ) ) {
+              carp_unique("Fast-path populate() with supplied related objects is not possible - falling back to regular create()");
+              return my $throwaway = $self->populate(@_);
+            }
+          }
 
+          push @$current_slice_seen_rel_infos, $rel_info->{$_};
+        }
+      }
 
-# populate() argumnets went over several incarnations
-# What we ultimately support is AoH
-sub _normalize_populate_args {
-  my ($self, $arg) = @_;
+      if ($current_slice_seen_rel_infos) {
+        push @$slices_with_rels, $data->[$i];
 
-  if (ref $arg eq 'ARRAY') {
-    if (!@$arg) {
-      return [];
+        # this is needed further down to decide whether or not to fallback to create()
+        $colinfo->{$_}{seen_null} ||= ! defined $data->[$i]{$_}
+          for keys %{$data->[$i]};
+      }
     }
-    elsif (ref $arg->[0] eq 'HASH') {
-      return $arg;
+    else {
+      $self->throw_exception('Unexpected populate() data structure member type: ' . ref $data->[$i] );
     }
-    elsif (ref $arg->[0] eq 'ARRAY') {
-      my @ret;
-      my @colnames = @{$arg->[0]};
-      foreach my $values (@{$arg}[1 .. $#$arg]) {
-        push @ret, { map { $colnames[$_] => $values->[$_] } (0 .. $#colnames) };
-      }
-      return \@ret;
+
+    if ( grep
+      { $_->{attrs}{is_depends_on} }
+      @{ $current_slice_seen_rel_infos || [] }
+    ) {
+      carp_unique("Fast-path populate() of belongs_to relationship data is not possible - falling back to regular create()");
+      return my $throwaway = $self->populate(@_);
     }
   }
 
-  $self->throw_exception('Populate expects an arrayref of hashrefs or arrayref of arrayrefs');
-}
+  if( $slices_with_rels ) {
 
-=head2 pager
+    # need to exclude the rel "columns"
+    $colnames = [ grep { ! $colinfo->{$_}{is_rel} } @$colnames ];
 
-=over 4
+    # extra sanity check - ensure the main source is in fact identifiable
+    # the localizing of nullability is insane, but oh well... the use-case is legit
+    my $ci = $rsrc->columns_info($colnames);
 
-=item Arguments: none
+    $ci->{$_} = { %{$ci->{$_}}, is_nullable => 0 }
+      for grep { ! $colinfo->{$_}{seen_null} } keys %$ci;
 
-=item Return Value: L<$pager|Data::Page>
+    unless( $rsrc->_identifying_column_set($ci) ) {
+      carp_unique("Fast-path populate() of non-uniquely identifiable rows with related data is not possible - falling back to regular create()");
+      return my $throwaway = $self->populate(@_);
+    }
+  }
+
+### inherit the data locked in the conditions of the resultset
+  my ($rs_data) = $self->_merge_with_rscond({});
+  delete @{$rs_data}{@$colnames};  # passed-in stuff takes precedence
+
+  # if anything left - decompose rs_data
+  my $rs_data_vals;
+  if (keys %$rs_data) {
+     push @$rs_data_vals, $rs_data->{$_}
+      for sort keys %$rs_data;
+  }
+
+### start work
+  $guard = $rsrc->schema->storage->txn_scope_guard
+    if $slices_with_rels;
+
+### main source data
+  # FIXME - need to switch entirely to a coderef-based thing,
+  # so that large sets aren't copied several times... I think
+  $rsrc->storage->insert_bulk(
+    $rsrc,
+    [ @$colnames, sort keys %$rs_data ],
+    [ map {
+      ref $data->[$_] eq 'ARRAY'
+      ? (
+          $slices_with_rels ? [ @{$data->[$_]}[0..$#$colnames], @{$rs_data_vals||[]} ]  # the collist changed
+        : $rs_data_vals     ? [ @{$data->[$_]}, @$rs_data_vals ]
+        :                     $data->[$_]
+      )
+      : [ @{$data->[$_]}{@$colnames}, @{$rs_data_vals||[]} ]
+    } $data_start .. $#$data ],
+  );
+
+### do the children relationships
+  if ( $slices_with_rels ) {
+    my @rels = grep { $colinfo->{$_}{is_rel} } keys %$colinfo
+      or die 'wtf... please report a bug with DBIC_TRACE=1 output (stacktrace)';
+
+    for my $sl (@$slices_with_rels) {
+
+      my ($main_proto, $main_proto_rs);
+      for my $rel (@rels) {
+        next unless defined $sl->{$rel};
+
+        $main_proto ||= {
+          %$rs_data,
+          (map { $_ => $sl->{$_} } @$colnames),
+        };
+
+        unless (defined $colinfo->{$rel}{rs}) {
+
+          $colinfo->{$rel}{rs} = $rsrc->related_source($rel)->resultset;
+
+          $colinfo->{$rel}{fk_map} = { reverse %{ $rsrc->_resolve_relationship_condition(
+            rel_name => $rel,
+            self_alias => "\xFE", # irrelevant
+            foreign_alias => "\xFF", # irrelevant
+          )->{identity_map} || {} } };
+
+        }
+
+        $colinfo->{$rel}{rs}->search({ map # only so that we inherit them values properly, no actual search
+          {
+            $_ => { '=' =>
+              ( $main_proto_rs ||= $rsrc->resultset->search($main_proto) )
+                ->get_column( $colinfo->{$rel}{fk_map}{$_} )
+                 ->as_query
+            }
+          }
+          keys %{$colinfo->{$rel}{fk_map}}
+        })->populate( ref $sl->{$rel} eq 'ARRAY' ? $sl->{$rel} : [ $sl->{$rel} ] );
+
+        1;
+      }
+    }
+  }
+
+  $guard->commit if $guard;
+}
+
+=head2 pager
+
+=over 4
+
+=item Arguments: none
+
+=item Return Value: L<$pager|Data::Page>
 
 =back
 
@@ -2272,7 +2540,7 @@ sub pager {
   # 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/;
+  delete @{$count_attrs}{qw/rows offset page pager/};
 
   my $total_rs = (ref $self)->new($self->result_source, $count_attrs);
 
@@ -2330,20 +2598,34 @@ sub new_result {
   $self->throw_exception( "new_result takes only one argument - a hashref of values" )
     if @_ > 2;
 
-  $self->throw_exception( "new_result expects a hashref" )
+  $self->throw_exception( "Result object instantiation requires a hashref as argument" )
     unless (ref $values eq 'HASH');
 
   my ($merged_cond, $cols_from_relations) = $self->_merge_with_rscond($values);
 
-  my %new = (
+  my $new = $self->result_class->new({
     %$merged_cond,
-    @$cols_from_relations
+    ( @$cols_from_relations
       ? (-cols_from_relations => $cols_from_relations)
-      : (),
+      : ()
+    ),
     -result_source => $self->result_source, # DO NOT REMOVE THIS, REQUIRED
-  );
+  });
+
+  if (
+    reftype($new) eq 'HASH'
+      and
+    ! keys %$new
+      and
+    blessed($new)
+  ) {
+    carp_unique (sprintf (
+      "%s->new returned a blessed empty hashref - a strong indicator something is wrong with its inheritance chain",
+      $self->result_class,
+    ));
+  }
 
-  return $self->result_class->new(\%new);
+  $new;
 }
 
 # _merge_with_rscond
@@ -2355,51 +2637,33 @@ sub new_result {
 sub _merge_with_rscond {
   my ($self, $data) = @_;
 
-  my (%new_data, @cols_from_relations);
+  my ($implied_data, @cols_from_relations);
 
   my $alias = $self->{attrs}{alias};
 
   if (! defined $self->{cond}) {
     # just massage $data below
   }
-  elsif ($self->{cond} eq $DBIx::Class::ResultSource::UNRESOLVABLE_CONDITION) {
-    %new_data = %{ $self->{attrs}{related_objects} || {} };  # nothing might have been inserted yet
-    @cols_from_relations = keys %new_data;
-  }
-  elsif (ref $self->{cond} ne 'HASH') {
-    $self->throw_exception(
-      "Can't abstract implicit construct, resultset condition not a hash"
-    );
+  elsif ($self->{cond} eq UNRESOLVABLE_CONDITION) {
+    $implied_data = $self->{attrs}{related_objects};  # nothing might have been inserted yet
+    @cols_from_relations = keys %{ $implied_data || {} };
   }
   else {
-    # precendence must be given to passed values over values inherited from
-    # the cond, so the order here is important.
-    my $collapsed_cond = $self->_collapse_cond($self->{cond});
-    my %implied = %{$self->_remove_alias($collapsed_cond, $alias)};
-
-    while ( my($col, $value) = each %implied ) {
-      my $vref = ref $value;
-      if (
-        $vref eq 'HASH'
-          and
-        keys(%$value) == 1
-          and
-        (keys %$value)[0] eq '='
-      ) {
-        $new_data{$col} = $value->{'='};
-      }
-      elsif( !$vref or $vref eq 'SCALAR' or blessed($value) ) {
-        $new_data{$col} = $value;
-      }
-    }
+    my $eqs = $self->result_source->schema->storage->_extract_fixed_condition_columns($self->{cond}, 'consider_nulls');
+    $implied_data = { map {
+      ( ($eqs->{$_}||'') eq UNRESOLVABLE_CONDITION ) ? () : ( $_ => $eqs->{$_} )
+    } keys %$eqs };
   }
 
-  %new_data = (
-    %new_data,
-    %{ $self->_remove_alias($data, $alias) },
+  return (
+    { map
+      { %{ $self->_remove_alias($_, $alias) } }
+      # precedence must be given to passed values over values inherited from
+      # the cond, so the order here is important.
+      ( $implied_data||(), $data)
+    },
+    \@cols_from_relations
   );
-
-  return (\%new_data, \@cols_from_relations);
 }
 
 # _has_resolved_attr
@@ -2407,7 +2671,7 @@ sub _merge_with_rscond {
 # determines if the resultset defines at least one
 # of the attributes supplied
 #
-# used to determine if a subquery is neccessary
+# used to determine if a subquery is necessary
 #
 # supports some virtual attributes:
 #   -join
@@ -2455,38 +2719,6 @@ sub _has_resolved_attr {
   return 0;
 }
 
-# _collapse_cond
-#
-# Recursively collapse the condition.
-
-sub _collapse_cond {
-  my ($self, $cond, $collapsed) = @_;
-
-  $collapsed ||= {};
-
-  if (ref $cond eq 'ARRAY') {
-    foreach my $subcond (@$cond) {
-      next unless ref $subcond;  # -or
-      $collapsed = $self->_collapse_cond($subcond, $collapsed);
-    }
-  }
-  elsif (ref $cond eq 'HASH') {
-    if (keys %$cond and (keys %$cond)[0] eq '-and') {
-      foreach my $subcond (@{$cond->{-and}}) {
-        $collapsed = $self->_collapse_cond($subcond, $collapsed);
-      }
-    }
-    else {
-      foreach my $col (keys %$cond) {
-        my $value = $cond->{$col};
-        $collapsed->{$col} = $value;
-      }
-    }
-  }
-
-  return $collapsed;
-}
-
 # _remove_alias
 #
 # Remove the specified alias from the specified query hash. A copy is made so
@@ -2531,16 +2763,11 @@ sub as_query {
 
   my $attrs = { %{ $self->_resolved_attrs } };
 
-  # For future use:
-  #
-  # in list ctx:
-  # my ($sql, \@bind, \%dbi_bind_attrs) = _select_args_to_query (...)
-  # $sql also has no wrapping parenthesis in list ctx
-  #
-  my $sqlbind = $self->result_source->storage
-    ->_select_args_to_query ($attrs->{from}, $attrs->{select}, $attrs->{where}, $attrs);
+  my $aq = $self->result_source->storage->_select_args_to_query (
+    $attrs->{from}, $attrs->{select}, $attrs->{where}, $attrs
+  );
 
-  return $sqlbind;
+  $aq;
 }
 
 =head2 find_or_new
@@ -2557,7 +2784,7 @@ sub as_query {
     { artist => 'fred' }, { key => 'artists' });
 
   $cd->cd_to_producer->find_or_new({ producer => $producer },
-                                   { key => 'primary });
+                                   { key => 'primary' });
 
 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
@@ -2672,10 +2899,9 @@ L</new>.
 =cut
 
 sub create {
-  my ($self, $attrs) = @_;
-  $self->throw_exception( "create needs a hashref" )
-    unless ref $attrs eq 'HASH';
-  return $self->new_result($attrs)->insert;
+  #my ($self, $col_data) = @_;
+  DBIx::Class::_ENV_::ASSERT_NO_INTERNAL_INDIRECT_CALLS and fail_on_internal_call;
+  return shift->new_result(shift)->insert;
 }
 
 =head2 find_or_create
@@ -2757,7 +2983,7 @@ sub find_or_create {
   if (keys %$hash and my $row = $self->find($hash, $attrs) ) {
     return $row;
   }
-  return $self->create($hash);
+  return $self->new_result($hash)->insert;
 }
 
 =head2 update_or_create
@@ -2827,7 +3053,7 @@ sub update_or_create {
     return $row;
   }
 
-  return $self->create($cond);
+  return $self->new_result($cond)->insert;
 }
 
 =head2 update_or_new
@@ -3012,8 +3238,10 @@ Returns a related resultset for the supplied relationship name.
 sub related_resultset {
   my ($self, $rel) = @_;
 
-  $self->{related_resultsets} ||= {};
-  return $self->{related_resultsets}{$rel} ||= do {
+  return $self->{related_resultsets}{$rel}
+    if defined $self->{related_resultsets}{$rel};
+
+  return $self->{related_resultsets}{$rel} = do {
     my $rsrc = $self->result_source;
     my $rel_info = $rsrc->relationship_info($rel);
 
@@ -3039,15 +3267,6 @@ sub related_resultset {
     #XXX - temp fix for result_class bug. There likely is a more elegant fix -groditi
     delete @{$attrs}{qw(result_class alias)};
 
-    my $new_cache;
-
-    if (my $cache = $self->get_cache) {
-      if ($cache->[0] && $cache->[0]->related_resultset($rel)->get_cache) {
-        $new_cache = [ map { @{$_->related_resultset($rel)->get_cache} }
-                        @$cache ];
-      }
-    }
-
     my $rel_source = $rsrc->related_source($rel);
 
     my $new = do {
@@ -3068,7 +3287,16 @@ sub related_resultset {
                        where => $attrs->{where},
                    });
     };
-    $new->set_cache($new_cache) if $new_cache;
+
+    if (my $cache = $self->get_cache) {
+      my @related_cache = map
+        { $_->related_resultset($rel)->get_cache || () }
+        @$cache
+      ;
+
+      $new->set_cache([ map @$_, @related_cache ]) if @related_cache == @$cache;
+    }
+
     $new;
   };
 }
@@ -3208,7 +3436,7 @@ sub _chain_relationship {
   # ->_resolve_join as otherwise they get lost - captainL
   my $join = $self->_merge_joinpref_attr( $attrs->{join}, $attrs->{prefetch} );
 
-  delete @{$attrs}{qw/join prefetch collapse group_by distinct select as columns +select +as +columns/};
+  delete @{$attrs}{qw/join prefetch collapse group_by distinct _grouped_by_distinct select as columns +select +as +columns/};
 
   my $seen = { %{ (delete $attrs->{seen_join}) || {} } };
 
@@ -3298,9 +3526,12 @@ sub _resolved_attrs {
   return $self->{_attrs} if $self->{_attrs};
 
   my $attrs  = { %{ $self->{attrs} || {} } };
-  my $source = $self->result_source;
+  my $source = $attrs->{result_source} = $self->result_source;
   my $alias  = $attrs->{alias};
 
+  $self->throw_exception("Specifying distinct => 1 in conjunction with collapse => 1 is unsupported")
+    if $attrs->{collapse} and $attrs->{distinct};
+
   # default selection list
   $attrs->{columns} = [ $source->columns ]
     unless List::Util::first { exists $attrs->{$_} } qw/columns cols select as/;
@@ -3338,14 +3569,10 @@ sub _resolved_attrs {
     if $attrs->{select};
 
   # assume all unqualified selectors to apply to the current alias (legacy stuff)
-  for (@sel) {
-    $_ = (ref $_ or $_ =~ /\./) ? $_ : "$alias.$_";
-  }
+  $_ = (ref $_ or $_ =~ /\./) ? $_ : "$alias.$_" for @sel;
 
-  # disqualify all $alias.col as-bits (collapser mandated)
-  for (@as) {
-    $_ = ($_ =~ /^\Q$alias.\E(.+)$/) ? $1 : $_;
-  }
+  # disqualify all $alias.col as-bits (inflate-map mandated)
+  $_ = ($_ =~ /^\Q$alias.\E(.+)$/) ? $1 : $_ for @as;
 
   # de-duplicate the result (remove *identical* select/as pairs)
   # and also die on duplicate {as} pointing to different {select}s
@@ -3415,32 +3642,21 @@ sub _resolved_attrs {
     $attrs->{group_by} = [ $attrs->{group_by} ];
   }
 
-  # generate the distinct induced group_by early, as prefetch will be carried via a
-  # subquery (since a group_by is present)
-  if (delete $attrs->{distinct}) {
-    if ($attrs->{group_by}) {
-      carp_unique ("Useless use of distinct on a grouped resultset ('distinct' is ignored when a 'group_by' is present)");
-    }
-    else {
-      # 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},
-      );
-    }
-  }
 
-  $attrs->{collapse} ||= {};
-  if ($attrs->{prefetch}) {
+  # generate selections based on the prefetch helper
+  my ($prefetch, @prefetch_select, @prefetch_as);
+  $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};
 
-    my $prefetch = $self->_merge_joinpref_attr( {}, delete $attrs->{prefetch} );
+    $self->throw_exception("Specifying prefetch in conjunction with an explicit collapse => 0 is unsupported")
+      if defined $attrs->{collapse} and ! $attrs->{collapse};
 
-    my $prefetch_ordering = [];
+    $attrs->{collapse} = 1;
 
     # this is a separate structure (we don't look in {from} directly)
     # as the resolver needs to shift things off the lists to work
@@ -3463,22 +3679,91 @@ sub _resolved_attrs {
       }
     }
 
-    my @prefetch =
-      $source->_resolve_prefetch( $prefetch, $alias, $join_map, $prefetch_ordering, $attrs->{collapse} );
+    my @prefetch = $source->_resolve_prefetch( $prefetch, $alias, $join_map );
+
+    # save these for after distinct resolution
+    @prefetch_select = map { $_->[0] } @prefetch;
+    @prefetch_as = map { $_->[1] } @prefetch;
+  }
+
+  # run through the resulting joinstructure (starting from our current slot)
+  # and unset collapse if proven unnecessary
+  #
+  # also while we are at it find out if the current root source has
+  # been premultiplied by previous related_source chaining
+  #
+  # this allows to predict whether a root object with all other relation
+  # data set to NULL is in fact unique
+  if ($attrs->{collapse}) {
+
+    if (ref $attrs->{from} eq 'ARRAY') {
+
+      if (@{$attrs->{from}} == 1) {
+        # no joins - no collapse
+        $attrs->{collapse} = 0;
+      }
+      else {
+        # find where our table-spec starts
+        my @fromlist = @{$attrs->{from}};
+        while (@fromlist) {
+          my $t = shift @fromlist;
+
+          my $is_multi;
+          # me vs join from-spec distinction - a ref means non-root
+          if (ref $t eq 'ARRAY') {
+            $t = $t->[0];
+            $is_multi ||= ! $t->{-is_single};
+          }
+          last if ($t->{-alias} && $t->{-alias} eq $alias);
+          $attrs->{_main_source_premultiplied} ||= $is_multi;
+        }
 
-    # we need to somehow mark which columns came from prefetch
-    if (@prefetch) {
-      my $sel_end = $#{$attrs->{select}};
-      $attrs->{_prefetch_selector_range} = [ $sel_end + 1, $sel_end + @prefetch ];
+        # no non-singles remaining, nor any premultiplication - nothing to collapse
+        if (
+          ! $attrs->{_main_source_premultiplied}
+            and
+          ! List::Util::first { ! $_->[0]{-is_single} } @fromlist
+        ) {
+          $attrs->{collapse} = 0;
+        }
+      }
     }
 
-    push @{ $attrs->{select} }, (map { $_->[0] } @prefetch);
-    push @{ $attrs->{as} }, (map { $_->[1] } @prefetch);
+    else {
+      # if we can not analyze the from - err on the side of safety
+      $attrs->{_main_source_premultiplied} = 1;
+    }
+  }
 
-    push( @{$attrs->{order_by}}, @$prefetch_ordering );
-    $attrs->{_collapse_order_by} = \@$prefetch_ordering;
+  # generate the distinct induced group_by before injecting the prefetched select/as parts
+  if (delete $attrs->{distinct}) {
+    if ($attrs->{group_by}) {
+      carp_unique ("Useless use of distinct on a grouped resultset ('distinct' is ignored when a 'group_by' is present)");
+    }
+    else {
+      $attrs->{_grouped_by_distinct} = 1;
+      # distinct affects only the main selection part, not what prefetch may add below
+      ($attrs->{group_by}, my $new_order) = $source->storage->_group_over_selection($attrs);
+
+      # FIXME possibly ignore a rewritten order_by (may turn out to be an issue)
+      # The thinking is: if we are collapsing the subquerying prefetch engine will
+      # rip stuff apart for us anyway, and we do not want to have a potentially
+      # function-converted external order_by
+      # ( there is an explicit if ( collapse && _grouped_by_distinct ) check in DBIHacks )
+      $attrs->{order_by} = $new_order unless $attrs->{collapse};
+    }
   }
 
+  # inject prefetch-bound selection (if any)
+  push @{$attrs->{select}}, @prefetch_select;
+  push @{$attrs->{as}}, @prefetch_as;
+
+  $attrs->{_simple_passthrough_construction} = !(
+    $attrs->{collapse}
+      or
+    grep { $_ =~ /\./ } @{$attrs->{as}}
+  );
+
   # 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
@@ -3603,7 +3888,7 @@ sub _merge_joinpref_attr {
     $seen_keys->{$import_key} = 1; # don't merge the same key twice
   }
 
-  return $orig;
+  return @$orig ? $orig : ();
 }
 
 {
@@ -3699,7 +3984,8 @@ sub STORABLE_freeze {
   my $to_serialize = { %$self };
 
   # A cursor in progress can't be serialized (and would make little sense anyway)
-  delete $to_serialize->{cursor};
+  # the parser can be regenerated (and can't be serialized)
+  delete @{$to_serialize}{qw/cursor _row_parser _result_inflator/};
 
   # 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') {
@@ -3736,6 +4022,10 @@ sub throw_exception {
   }
 }
 
+1;
+
+__END__
+
 # XXX: FIXME: Attributes docs need clearing up
 
 =head1 ATTRIBUTES
@@ -3785,39 +4075,57 @@ syntax as outlined above.
 
 =over 4
 
-=item Value: \@columns
+=item Value: \@columns | \%columns | $column
 
 =back
 
 Shortcut to request a particular set of columns to be retrieved. Each
 column spec may be a string (a table column name), or a hash (in which
 case the key is the C<as> value, and the value is used as the C<select>
-expression). Adds C<me.> onto the start of any column without a C<.> in
+expression). Adds the L</current_source_alias> onto the start of any column without a C<.> in
 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.)
+earlier versions of DBIC, but this is deprecated)
 
 Essentially C<columns> does the same as L</select> and L</as>.
 
-    columns => [ 'foo', { bar => 'baz' } ]
+    columns => [ 'some_column', { dbic_slot => 'another_column' } ]
 
 is the same as
 
-    select => [qw/foo baz/],
-    as => [qw/foo bar/]
+    select => [qw(some_column another_column)],
+    as     => [qw(some_column dbic_slot)]
+
+If you want to individually retrieve related columns (in essence perform
+manual prefetch) you have to make sure to specify the correct inflation slot
+chain such that it matches existing relationships:
+
+    my $rs = $schema->resultset('Artist')->search({}, {
+        # required to tell DBIC to collapse has_many relationships
+        collapse => 1,
+        join     => { cds => 'tracks'},
+        '+columns'  => {
+          'cds.cdid'         => 'cds.cdid',
+          'cds.tracks.title' => 'tracks.title',
+        },
+    });
 
 =head2 +columns
 
+B<NOTE:> You B<MUST> explicitly quote C<'+columns'> when using this attribute.
+Not doing so causes Perl to incorrectly interpret C<+columns> as a bareword
+with a unary plus operator before it, which is the same as simply C<columns>.
+
 =over 4
 
-=item Value: \@columns
+=item Value: \@extra_columns
 
 =back
 
-Indicates additional columns to be selected from storage. Works the same
-as L</columns> but adds columns to the selection. (You may also use the
-C<include_columns> attribute, as in earlier versions of DBIC). For
-example:-
+Indicates additional columns to be selected from storage. Works the same as
+L</columns> but adds columns to the current selection. (You may also use the
+C<include_columns> attribute, as in earlier versions of DBIC, but this is
+deprecated)
 
   $schema->resultset('CD')->search(undef, {
     '+columns' => ['artist.name'],
@@ -3829,20 +4137,6 @@ 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
-
-=item Value: \@columns
-
-=back
-
-Deprecated.  Acts as a synonym for L</+columns> for backward compatibility.
-
 =head2 select
 
 =over 4
@@ -3873,28 +4167,22 @@ 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 default selection, instead of specifying
-an explicit list.
-
-=back
-
-=head2 +as
+B<NOTE:> You B<MUST> explicitly quote C<'+select'> when using this attribute.
+Not doing so causes Perl to incorrectly interpret C<+select> as a bareword
+with a unary plus operator before it, which is the same as simply C<select>.
 
 =over 4
 
-Indicates additional column names for those added via L</+select>. See L</as>.
+=item Value: \@extra_select_columns
 
 =back
 
+Indicates additional columns to be selected from storage.  Works the same as
+L</select> but adds columns to the current selection, instead of specifying
+a new explicit list.
+
 =head2 as
 
 =over 4
@@ -3903,7 +4191,7 @@ Indicates additional column names for those added via L</+select>. See L</as>.
 
 =back
 
-Indicates column names for object inflation. That is L</as> indicates the
+Indicates DBIC-side 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
@@ -3937,6 +4225,20 @@ use C<get_column> instead:
 You can create your own accessors if required - see
 L<DBIx::Class::Manual::Cookbook> for details.
 
+=head2 +as
+
+B<NOTE:> You B<MUST> explicitly quote C<'+as'> when using this attribute.
+Not doing so causes Perl to incorrectly interpret C<+as> as a bareword
+with a unary plus operator before it, which is the same as simply C<as>.
+
+=over 4
+
+=item Value: \@extra_inflation_names
+
+=back
+
+Indicates additional inflation names for selectors added via L</+select>. See L</as>.
+
 =head2 join
 
 =over 4
@@ -4000,7 +4302,7 @@ similarly for a third time). For e.g.
 will return a set of all artists that have both a cd with title 'Down
 to Earth' and a cd with title 'Popular'.
 
-If you want to fetch related objects from other tables as well, see C<prefetch>
+If you want to fetch related objects from other tables as well, see L</prefetch>
 below.
 
  NOTE: An internal join-chain pruner will discard certain joins while
@@ -4011,185 +4313,133 @@ below.
 
 For more help on using joins with search, see L<DBIx::Class::Manual::Joining>.
 
-=head2 prefetch
+=head2 collapse
 
 =over 4
 
-=item Value: ($rel_name | \@rel_names | \%rel_names)
+=item Value: (0 | 1)
 
 =back
 
-Contains one or more relationships that should be fetched along with
-the main query (when they are accessed afterwards the data will
-already be available, without extra queries to the database).  This is
-useful for when you know you will need the related objects, because it
-saves at least one query:
-
-  my $rs = $schema->resultset('Tag')->search(
-    undef,
-    {
-      prefetch => {
-        cd => 'artist'
-      }
-    }
-  );
-
-The initial search results in SQL like the following:
-
-  SELECT tag.*, cd.*, artist.* FROM tag
-  JOIN cd ON tag.cd = cd.cdid
-  JOIN artist ON cd.artist = artist.artistid
-
-L<DBIx::Class> has no need to go back to the database when we access the
-C<cd> or C<artist> relationships, which saves us two SQL statements in this
-case.
-
-Simple prefetches will be joined automatically, so there is no need
-for a C<join> attribute in the above search.
-
-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.
-
- # 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 => [
-       'tracks',                         # has_many
-       { cd_to_producer => 'producer' }, # has_many => belongs_to (i.e. m2m)
-     ]
-   }
- );
+When set to a true value, indicates that any rows fetched from joined has_many
+relationships are to be aggregated into the corresponding "parent" object. For
+example, the resultset:
 
-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({}, {
+    '+columns' => [ qw/ tracks.title tracks.position / ],
+    join => 'tracks',
+    collapse => 1,
+  });
 
-  my $rs = $schema->resultset('CD')->search(
-   {'record_label.name' => 'Music Product Ltd.'},
-   {
-     join     => {artist => 'record_label'},
-     prefetch => 'artist',
-   }
- );
+While executing the following query:
 
-... will work, searching on the record label's name, but only
-prefetching the C<artist>.
+  SELECT me.*, tracks.title, tracks.position
+    FROM cd me
+    LEFT JOIN track tracks
+      ON tracks.cdid = me.cdid
 
-=head3 Using L</prefetch> with L</select> / L</+select> / L</as> / L</+as>
+Will return only as many objects as there are rows in the CD source, even
+though the result of the query may span many rows. Each of these CD objects
+will in turn have multiple "Track" objects hidden behind the has_many
+generated accessor C<tracks>. Without C<< collapse => 1 >>, the return values
+of this resultset would be as many CD objects as there are tracks (a "Cartesian
+product"), with each CD object containing exactly one of all fetched Track data.
 
-L</prefetch> implies a L</+select>/L</+as> with the fields of the
-prefetched relations.  So given:
+When a collapse is requested on a non-ordered resultset, an order by some
+unique part of the main source (the left-most table) is inserted automatically.
+This is done so that the resultset is allowed to be "lazy" - calling
+L<< $rs->next|/next >> will fetch only as many rows as it needs to build the next
+object with all of its related data.
 
-  my $rs = $schema->resultset('CD')->search(
-   undef,
-   {
-     select   => ['cd.title'],
-     as       => ['cd_title'],
-     prefetch => 'artist',
-   }
- );
+If an L</order_by> is already declared, and orders the resultset in a way that
+makes collapsing as described above impossible (e.g. C<< ORDER BY
+has_many_rel.column >> or C<ORDER BY RANDOM()>), DBIC will automatically
+switch to "eager" mode and slurp the entire resultset before constructing the
+first object returned by L</next>.
 
-The L</select> becomes: C<'cd.title', 'artist.*'> and the L</as>
-becomes: C<'cd_title', 'artist.*'>.
+Setting this attribute on a resultset that does not join any has_many
+relations is a no-op.
 
-=head3 CAVEATS
+For a more in-depth discussion, see L</PREFETCHING>.
 
-Prefetch does a lot of deep magic. As such, it may not behave exactly
-as you might expect.
+=head2 prefetch
 
 =over 4
 
-=item *
-
-Prefetch uses the L</cache> to populate the prefetched relationships. This
-may or may not be what you want.
+=item Value: ($rel_name | \@rel_names | \%rel_names)
 
-=item *
+=back
 
-If you specify a condition on a prefetched relationship, ONLY those
-rows that match the prefetched condition will be fetched into that relationship.
-This means that adding prefetch to a search() B<may alter> what is returned by
-traversing a relationship. So, if you have C<< Artist->has_many(CDs) >> and you do
+This attribute is a shorthand for specifying a L</join> spec, adding all
+columns from the joined related sources as L</+columns> and setting
+L</collapse> to a true value. For example, the following two queries are
+equivalent:
 
-  my $artist_rs = $schema->resultset('Artist')->search({
-      'cds.year' => 2008,
-  }, {
-      join => 'cds',
+  my $rs = $schema->resultset('Artist')->search({}, {
+    prefetch => { cds => ['genre', 'tracks' ] },
   });
 
-  my $count = $artist_rs->first->cds->count;
+and
 
-  my $artist_rs_prefetch = $artist_rs->search( {}, { prefetch => 'cds' } );
+  my $rs = $schema->resultset('Artist')->search({}, {
+    join => { cds => ['genre', 'tracks' ] },
+    collapse => 1,
+    '+columns' => [
+      (map
+        { +{ "cds.$_" => "cds.$_" } }
+        $schema->source('Artist')->related_source('cds')->columns
+      ),
+      (map
+        { +{ "cds.genre.$_" => "genre.$_" } }
+        $schema->source('Artist')->related_source('cds')->related_source('genre')->columns
+      ),
+      (map
+        { +{ "cds.tracks.$_" => "tracks.$_" } }
+        $schema->source('Artist')->related_source('cds')->related_source('tracks')->columns
+      ),
+    ],
+  });
 
-  my $prefetch_count = $artist_rs_prefetch->first->cds->count;
+Both producing the following SQL:
+
+  SELECT  me.artistid, me.name, me.rank, me.charfield,
+          cds.cdid, cds.artist, cds.title, cds.year, cds.genreid, cds.single_track,
+          genre.genreid, genre.name,
+          tracks.trackid, tracks.cd, tracks.position, tracks.title, tracks.last_updated_on, tracks.last_updated_at
+    FROM artist me
+    LEFT JOIN cd cds
+      ON cds.artist = me.artistid
+    LEFT JOIN genre genre
+      ON genre.genreid = cds.genreid
+    LEFT JOIN track tracks
+      ON tracks.cd = cds.cdid
+  ORDER BY me.artistid
+
+While L</prefetch> implies a L</join>, it is ok to mix the two together, as
+the arguments are properly merged and generally do the right thing. For
+example, you may want to do the following:
+
+  my $artists_and_cds_without_genre = $schema->resultset('Artist')->search(
+    { 'genre.genreid' => undef },
+    {
+      join => { cds => 'genre' },
+      prefetch => 'cds',
+    }
+  );
 
-  cmp_ok( $count, '==', $prefetch_count, "Counts should be the same" );
+Which generates the following SQL:
 
-that cmp_ok() may or may not pass depending on the datasets involved. This
-behavior may or may not survive the 0.09 transition.
+  SELECT  me.artistid, me.name, me.rank, me.charfield,
+          cds.cdid, cds.artist, cds.title, cds.year, cds.genreid, cds.single_track
+    FROM artist me
+    LEFT JOIN cd cds
+      ON cds.artist = me.artistid
+    LEFT JOIN genre genre
+      ON genre.genreid = cds.genreid
+  WHERE genre.genreid IS NULL
+  ORDER BY me.artistid
 
-=back
+For a more in-depth discussion, see L</PREFETCHING>.
 
 =head2 alias
 
@@ -4318,8 +4568,17 @@ or with an in-place function in which case literal SQL is required:
 
 =back
 
-Set to 1 to group by all columns. If the resultset already has a group_by
-attribute, this setting is ignored and an appropriate warning is issued.
+Set to 1 to automatically generate a L</group_by> clause based on the selection
+(including intelligent handling of L</order_by> contents). Note that the group
+criteria calculation takes place over the B<final> selection. This includes
+any L</+columns>, L</+select> or L</order_by> additions in subsequent
+L</search> calls, and standalone columns selected via
+L<DBIx::Class::ResultSetColumn> (L</get_column>). A notable exception are the
+extra selections specified via L</prefetch> - such selections are explicitly
+excluded from group criteria calculations.
+
+If the final ResultSet also explicitly defines a L</group_by> attribute, this
+setting is ignored and an appropriate warning is issued.
 
 =head2 where
 
@@ -4367,6 +4626,131 @@ Set to 'update' for a SELECT ... FOR UPDATE or 'shared' for a SELECT
 ... FOR SHARED. If \$scalar is passed, this is taken directly and embedded in the
 query.
 
+=head1 PREFETCHING
+
+DBIx::Class supports arbitrary related data prefetching from multiple related
+sources. Any combination of relationship types and column sets are supported.
+If L<collapsing|/collapse> is requested, there is an additional requirement of
+selecting enough data to make every individual object uniquely identifiable.
+
+Here are some more involved examples, based on the following relationship map:
+
+  # Assuming:
+  My::Schema::CD->belongs_to( artist      => 'My::Schema::Artist'     );
+  My::Schema::CD->might_have( liner_note  => 'My::Schema::LinerNotes' );
+  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('Tag')->search(
+    undef,
+    {
+      prefetch => {
+        cd => 'artist'
+      }
+    }
+  );
+
+The initial search results in SQL like the following:
+
+  SELECT tag.*, cd.*, artist.* FROM tag
+  JOIN cd ON tag.cd = cd.cdid
+  JOIN artist ON cd.artist = artist.artistid
+
+L<DBIx::Class> has no need to go back to the database when we access the
+C<cd> or C<artist> relationships, which saves us two SQL statements in this
+case.
+
+Simple prefetches will be joined automatically, so there is no need
+for a C<join> attribute in the above search.
+
+The L</prefetch> attribute can be used with 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('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.
+
+=head3 CAVEATS
+
+Prefetch does a lot of deep magic. As such, it may not behave exactly
+as you might expect.
+
+=over 4
+
+=item *
+
+Prefetch uses the L</cache> to populate the prefetched relationships. This
+may or may not be what you want.
+
+=item *
+
+If you specify a condition on a prefetched relationship, ONLY those
+rows that match the prefetched condition will be fetched into that relationship.
+This means that adding prefetch to a search() B<may alter> what is returned by
+traversing a relationship. So, if you have C<< Artist->has_many(CDs) >> and you do
+
+  my $artist_rs = $schema->resultset('Artist')->search({
+      'cds.year' => 2008,
+  }, {
+      join => 'cds',
+  });
+
+  my $count = $artist_rs->first->cds->count;
+
+  my $artist_rs_prefetch = $artist_rs->search( {}, { prefetch => 'cds' } );
+
+  my $prefetch_count = $artist_rs_prefetch->first->cds->count;
+
+  cmp_ok( $count, '==', $prefetch_count, "Counts should be the same" );
+
+That cmp_ok() may or may not pass depending on the datasets involved. In other
+words the C<WHERE> condition would apply to the entire dataset, just like
+it would in regular SQL. If you want to add a condition only to the "right side"
+of a C<LEFT JOIN> - consider declaring and using a L<relationship with a custom
+condition|DBIx::Class::Relationship::Base/condition>
+
+=back
+
 =head1 DBIC BIND VALUES
 
 Because DBIC may need more information to bind values than just the column name
@@ -4403,7 +4787,7 @@ or to a sensible value based on the "data_type".
 =item dbic_colname
 
 Used to fill in missing sqlt_datatype and sqlt_size attributes (if they are
-explicitly specified they are never overriden).  Also used by some weird DBDs,
+explicitly specified they are never overridden).  Also used by some weird DBDs,
 where the column name should be available at bind_param time (e.g. Oracle).
 
 =back
@@ -4414,6 +4798,7 @@ supported:
   [ $name => $val ] === [ { dbic_colname => $name }, $val ]
   [ \$dt  => $val ] === [ { sqlt_datatype => $dt }, $val ]
   [ undef,   $val ] === [ {}, $val ]
+  $val              === [ {}, $val ]
 
 =head1 AUTHOR AND CONTRIBUTORS
 
@@ -4423,6 +4808,3 @@ See L<AUTHOR|DBIx::Class/AUTHOR> and L<CONTRIBUTORS|DBIx::Class/CONTRIBUTORS> in
 
 You may distribute this code under the same terms as Perl itself.
 
-=cut
-
-1;