X-Git-Url: http://git.shadowcat.co.uk/gitweb/gitweb.cgi?a=blobdiff_plain;f=lib%2FDBIx%2FClass%2FResultSet.pm;h=4f7dd00e346115ea6c7995a52985a7c287cc515c;hb=9c3018b6b589a8250b75a3db451a5baac9c6c3df;hp=e77108b4aadedd600a0ba55c2cfdf01703506d76;hpb=6f8ab6dc1902d7536b682bd6c32c22b48f3e5e3d;p=dbsrgits%2FDBIx-Class.git

diff --git a/lib/DBIx/Class/ResultSet.pm b/lib/DBIx/Class/ResultSet.pm
index e77108b..4f7dd00 100644
--- a/lib/DBIx/Class/ResultSet.pm
+++ b/lib/DBIx/Class/ResultSet.pm
@@ -3,52 +3,146 @@ package DBIx::Class::ResultSet;
 use strict;
 use warnings;
 use overload
-        '0+'     => \&count,
-        'bool'   => sub { 1; },
+        '0+'     => "count",
+        'bool'   => "_bool",
         fallback => 1;
 use Carp::Clan qw/^DBIx::Class/;
 use Data::Page;
 use Storable;
 use DBIx::Class::ResultSetColumn;
 use DBIx::Class::ResultSourceHandle;
+use List::Util ();
+use Scalar::Util ();
 use base qw/DBIx::Class/;
 
-__PACKAGE__->mk_group_accessors('simple' => qw/result_class _source_handle/);
+__PACKAGE__->mk_group_accessors('simple' => qw/_result_class _source_handle/);
 
 =head1 NAME
 
-DBIx::Class::ResultSet - Responsible for fetching and creating resultset.
+DBIx::Class::ResultSet - Represents a query used for fetching a set of results.
 
 =head1 SYNOPSIS
 
-  my $rs   = $schema->resultset('User')->search(registered => 1);
-  my @rows = $schema->resultset('CD')->search(year => 2005);
+  my $users_rs   = $schema->resultset('User');
+  my $registered_users_rs   = $schema->resultset('User')->search({ registered => 1 });
+  my @cds_in_2005 = $schema->resultset('CD')->search({ year => 2005 })->all();
 
 =head1 DESCRIPTION
 
-The resultset is also known as an iterator. It is responsible for handling
-queries that may return an arbitrary number of rows, e.g. via L</search>
-or a C<has_many> relationship.
+A ResultSet is an object which stores a set of conditions representing
+a query. It is the backbone of DBIx::Class (i.e. the really
+important/useful bit).
 
-In the examples below, the following table classes are used:
+No SQL is executed on the database when a ResultSet is created, it
+just stores all the conditions needed to create the query.
 
-  package MyApp::Schema::Artist;
-  use base qw/DBIx::Class/;
-  __PACKAGE__->load_components(qw/Core/);
-  __PACKAGE__->table('artist');
-  __PACKAGE__->add_columns(qw/artistid name/);
-  __PACKAGE__->set_primary_key('artistid');
-  __PACKAGE__->has_many(cds => 'MyApp::Schema::CD');
-  1;
+A basic ResultSet representing the data of an entire table is returned
+by calling C<resultset> on a L<DBIx::Class::Schema> and passing in a
+L<Source|DBIx::Class::Manual::Glossary/Source> name.
 
-  package MyApp::Schema::CD;
-  use base qw/DBIx::Class/;
-  __PACKAGE__->load_components(qw/Core/);
-  __PACKAGE__->table('cd');
-  __PACKAGE__->add_columns(qw/cdid artist title year/);
-  __PACKAGE__->set_primary_key('cdid');
-  __PACKAGE__->belongs_to(artist => 'MyApp::Schema::Artist');
-  1;
+  my $users_rs = $schema->resultset('User');
+
+A new ResultSet is returned from calling L</search> on an existing
+ResultSet. The new one will contain all the conditions of the
+original, plus any new conditions added in the C<search> call.
+
+A ResultSet also incorporates an implicit iterator. L</next> and L</reset>
+can be used to walk through all the L<DBIx::Class::Row>s the ResultSet
+represents.
+
+The query that the ResultSet represents is B<only> executed against
+the database when these methods are called:
+L</find> L</next> L</all> L</first> L</single> L</count>
+
+=head1 EXAMPLES
+
+=head2 Chaining resultsets
+
+Let's say you've got a query that needs to be run to return some data
+to the user. But, you have an authorization system in place that
+prevents certain users from seeing certain information. So, you want
+to construct the basic query in one method, but add constraints to it in
+another.
+
+  sub get_data {
+    my $self = shift;
+    my $request = $self->get_request; # Get a request object somehow.
+    my $schema = $self->get_schema;   # Get the DBIC schema object somehow.
+
+    my $cd_rs = $schema->resultset('CD')->search({
+      title => $request->param('title'),
+      year => $request->param('year'),
+    });
+
+    $self->apply_security_policy( $cd_rs );
+
+    return $cd_rs->all();
+  }
+
+  sub apply_security_policy {
+    my $self = shift;
+    my ($rs) = @_;
+
+    return $rs->search({
+      subversive => 0,
+    });
+  }
+
+=head3 Resolving conditions and attributes
+
+When a resultset is chained from another resultset, conditions and
+attributes with the same keys need resolving.
+
+L</join>, L</prefetch>, L</+select>, L</+as> attributes are merged
+into the existing ones from the original resultset.
+
+The L</where>, L</having> attribute, and any search conditions are
+merged with an SQL C<AND> to the existing condition from the original
+resultset.
+
+All other attributes are overridden by any new ones supplied in the
+search attributes.
+
+=head2 Multiple queries
+
+Since a resultset just defines a query, you can do all sorts of
+things with it with the same object.
+
+  # Don't hit the DB yet.
+  my $cd_rs = $schema->resultset('CD')->search({
+    title => 'something',
+    year => 2009,
+  });
+
+  # Each of these hits the DB individually.
+  my $count = $cd_rs->count;
+  my $most_recent = $cd_rs->get_column('date_released')->max();
+  my @records = $cd_rs->all;
+
+And it's not just limited to SELECT statements.
+
+  $cd_rs->delete();
+
+This is even cooler:
+
+  $cd_rs->create({ artist => 'Fred' });
+
+Which is the same as:
+
+  $schema->resultset('CD')->create({
+    title => 'something',
+    year => 2009,
+    artist => 'Fred'
+  });
+
+See: L</search>, L</count>, L</get_column>, L</all>, L</create>.
+
+=head1 OVERLOADING
+
+If a resultset is used in a numeric context it returns the L</count>.
+However, if it is used in a booleand context it is always true.  So if
+you want to check if a resultset has any results use C<if $rs != 0>.
+C<if $rs> will always be true.
 
 =head1 METHODS
 
@@ -85,21 +179,20 @@ sub new {
   return $class->new_result(@_) if ref $class;
 
   my ($source, $attrs) = @_;
-  $source = $source->handle 
+  $source = $source->handle
     unless $source->isa('DBIx::Class::ResultSourceHandle');
   $attrs = { %{$attrs||{}} };
 
   if ($attrs->{page}) {
     $attrs->{rows} ||= 10;
-    $attrs->{offset} ||= 0;
-    $attrs->{offset} += ($attrs->{rows} * ($attrs->{page} - 1));
   }
 
   $attrs->{alias} ||= 'me';
 
+  # Creation of {} and bless separated to mitigate RH perl bug
+  # see https://bugzilla.redhat.com/show_bug.cgi?id=196836
   my $self = {
     _source_handle => $source,
-    result_class => $attrs->{result_class} || $source->resolve->result_class,
     cond => $attrs->{where},
     count => undef,
     pager => undef,
@@ -108,6 +201,10 @@ sub new {
 
   bless $self, $class;
 
+  $self->result_class(
+    $attrs->{result_class} || $source->resolve->result_class
+  );
+
   return $self;
 }
 
@@ -140,6 +237,8 @@ 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>.
 
+For more help on using joins with search, see L<DBIx::Class::Manual::Joining>.
+
 =cut
 
 sub search {
@@ -166,10 +265,9 @@ always return a resultset, even in list context.
 sub search_rs {
   my $self = shift;
 
-  my $rows;
-
-  unless (@_) {                 # no search, effectively just a clone
-    $rows = $self->get_cache;
+  # Special-case handling for (undef, undef).
+  if ( @_ == 2 && !defined $_[1] && !defined $_[0] ) {
+    pop(@_); pop(@_);
   }
 
   my $attrs = {};
@@ -178,10 +276,24 @@ sub search_rs {
   my $having = delete $our_attrs->{having};
   my $where = delete $our_attrs->{where};
 
+  my $rows;
+
+  my %safe = (alias => 1, cache => 1);
+
+  unless (
+    (@_ && defined($_[0])) # @_ == () or (undef)
+    ||
+    (keys %$attrs # empty attrs or only 'safe' attrs
+    && List::Util::first { !$safe{$_} } keys %$attrs)
+  ) {
+    # no search, effectively just a clone
+    $rows = $self->get_cache;
+  }
+
   my $new_attrs = { %{$our_attrs}, %{$attrs} };
 
   # merge new attrs into inherited
-  foreach my $key (qw/join prefetch/) {
+  foreach my $key (qw/join prefetch +select +as bind/) {
     next unless exists $attrs->{$key};
     $new_attrs->{$key} = $self->_merge_attr($our_attrs->{$key}, $attrs->{$key});
   }
@@ -266,13 +378,30 @@ sub search_rs {
 Pass a literal chunk of SQL to be added to the conditional part of the
 resultset query.
 
+CAVEAT: C<search_literal> is provided for Class::DBI compatibility and should
+only be used in that context. C<search_literal> is a convenience method.
+It is equivalent to calling $schema->search(\[]), but if you want to ensure
+columns are bound correctly, use C<search>.
+
+Example of how to use C<search> instead of C<search_literal>
+
+  my @cds = $cd_rs->search_literal('cdid = ? AND (artist = ? OR artist = ?)', (2, 1, 2));
+  my @cds = $cd_rs->search(\[ 'cdid = ? AND (artist = ? OR artist = ?)', [ 'cdid', 2 ], [ 'artist', 1 ], [ 'artist', 2 ] ]);
+
+
+See L<DBIx::Class::Manual::Cookbook/Searching> and
+L<DBIx::Class::Manual::FAQ/Searching> for searching techniques that do not
+require C<search_literal>.
+
 =cut
 
 sub search_literal {
-  my ($self, $cond, @vals) = @_;
-  my $attrs = (ref $vals[$#vals] eq 'HASH' ? { %{ pop(@vals) } } : {});
-  $attrs->{bind} = [ @{$self->{attrs}{bind}||[]}, @vals ];
-  return $self->search(\$cond, $attrs);
+  my ($self, $sql, @bind) = @_;
+  my $attr;
+  if ( @bind && ref($bind[-1]) eq 'HASH' ) {
+    $attr = pop @bind;
+  }
+  return $self->search(\[ $sql, map [ __DUMMY__ => $_ ], @bind ], ($attr || () ));
 }
 
 =head2 find
@@ -281,7 +410,7 @@ sub search_literal {
 
 =item Arguments: @values | \%cols, \%attrs?
 
-=item Return Value: $row_object
+=item Return Value: $row_object | undef
 
 =back
 
@@ -310,11 +439,18 @@ Additionally, you can specify the columns explicitly by name:
 If the C<key> is specified as C<primary>, it searches only on the primary key.
 
 If no C<key> is specified, it searches on all unique constraints defined on the
-source, including the primary key.
+source for which column data is provided, including the primary key.
 
 If your table does not have a primary key, you B<must> provide a value for the
 C<key> attribute matching one of the unique constraints on the source.
 
+In addition to C<key>, L</find> recognizes and applies standard
+L<resultset attributes|/ATTRIBUTES> in the same way as L</search> does.
+
+Note: If your query does not return only one row, a warning is generated:
+
+  Query returned more than one row
+
 See also L</find_or_create> and L</update_or_create>. For information on how to
 declare unique constraints, see
 L<DBIx::Class::ResultSource/add_unique_constraint>.
@@ -355,7 +491,7 @@ sub find {
         && ($info = $self->result_source->relationship_info($key))) {
       my $val = delete $input_query->{$key};
       next KEY if (ref($val) eq 'ARRAY'); # has_many for multi_create
-      my $rel_q = $self->result_source->resolve_condition(
+      my $rel_q = $self->result_source->_resolve_condition(
                     $info->{cond}, $val, $key
                   );
       die "Can't handle OR join condition in find" if ref($rel_q) eq 'ARRAY';
@@ -366,25 +502,41 @@ sub find {
     @{$input_query}{@keys} = values %related;
   }
 
-  my @unique_queries = $self->_unique_queries($input_query, $attrs);
 
   # Build the final query: Default to the disjunction of the unique queries,
   # but allow the input query in case the ResultSet defines the query or the
   # user is abusing find
   my $alias = exists $attrs->{alias} ? $attrs->{alias} : $self->{attrs}{alias};
-  my $query = @unique_queries
-    ? [ map { $self->_add_alias($_, $alias) } @unique_queries ]
-    : $self->_add_alias($input_query, $alias);
+  my $query;
+  if (exists $attrs->{key}) {
+    my @unique_cols = $self->result_source->unique_constraint_columns($attrs->{key});
+    my $unique_query = $self->_build_unique_query($input_query, \@unique_cols);
+    $query = $self->_add_alias($unique_query, $alias);
+  }
+  elsif ($self->{attrs}{accessor} and $self->{attrs}{accessor} eq 'single') {
+    # This means that we got here after a merger of relationship conditions
+    # in ::Relationship::Base::search_related (the row method), and furthermore
+    # the relationship is of the 'single' type. This means that the condition
+    # provided by the relationship (already attached to $self) is sufficient,
+    # as there can be only one row in the databse that would satisfy the 
+    # relationship
+  }
+  else {
+    my @unique_queries = $self->_unique_queries($input_query, $attrs);
+    $query = @unique_queries
+      ? [ map { $self->_add_alias($_, $alias) } @unique_queries ]
+      : $self->_add_alias($input_query, $alias);
+  }
 
   # Run the query
-  if (keys %$attrs) {
-    my $rs = $self->search($query, $attrs);
-    return keys %{$rs->_resolved_attrs->{collapse}} ? $rs->next : $rs->single;
+  my $rs = $self->search ($query, $attrs);
+  if (keys %{$rs->_resolved_attrs->{collapse}}) {
+    my $row = $rs->next;
+    carp "Query returned more than one row" if $rs->next;
+    return $row;
   }
   else {
-    return keys %{$self->_resolved_attrs->{collapse}}
-      ? $self->search($query)->next
-      : $self->single($query);
+    return $rs->single;
   }
 }
 
@@ -474,6 +626,17 @@ sub search_related {
   return shift->related_resultset(shift)->search(@_);
 }
 
+=head2 search_related_rs
+
+This method works exactly the same as search_related, except that
+it guarantees a restultset, even in list context.
+
+=cut
+
+sub search_related_rs {
+  return shift->related_resultset(shift)->search_rs(@_);
+}
+
 =head2 cursor
 
 =over 4
@@ -492,7 +655,8 @@ L<DBIx::Class::Cursor> for more information.
 sub cursor {
   my ($self) = @_;
 
-  my $attrs = { %{$self->_resolved_attrs} };
+  my $attrs = $self->_resolved_attrs_copy;
+
   return $self->{cursor}
     ||= $self->result_source->storage->select($attrs->{from}, $attrs->{select},
           $attrs->{where},$attrs);
@@ -511,17 +675,50 @@ sub cursor {
   my $cd = $schema->resultset('CD')->single({ year => 2001 });
 
 Inflates the first result without creating a cursor if the resultset has
-any records in it; if not returns nothing. Used by L</find> as an optimisation.
+any records in it; if not returns nothing. Used by L</find> as a lean version of
+L</search>.
+
+While this method can take an optional search condition (just like L</search>)
+being a fast-code-path it does not recognize search attributes. If you need to
+add extra joins or similar, call L</search> and then chain-call L</single> on the
+L<DBIx::Class::ResultSet> returned.
+
+=over
+
+=item B<Note>
+
+As of 0.08100, this method enforces the assumption that the preceeding
+query returns only one row. If more than one row is returned, you will receive
+a warning:
 
-Can optionally take an additional condition *only* - this is a fast-code-path
-method; if you need to add extra joins or similar call ->search and then
-->single without a condition on the $rs returned from that.
+  Query returned more than one row
+
+In this case, you should be using L</next> or L</find> instead, or if you really
+know what you are doing, use the L</rows> attribute to explicitly limit the size
+of the resultset.
+
+This method will also throw an exception if it is called on a resultset prefetching
+has_many, as such a prefetch implies fetching multiple rows from the database in
+order to assemble the resulting object.
+
+=back
 
 =cut
 
 sub single {
   my ($self, $where) = @_;
-  my $attrs = { %{$self->_resolved_attrs} };
+  if(@_ > 2) {
+      $self->throw_exception('single() only takes search conditions, no attributes. You want ->search( $cond, $attrs )->single()');
+  }
+
+  my $attrs = $self->_resolved_attrs_copy;
+
+  if (keys %{$attrs->{collapse}}) {
+    $self->throw_exception(
+      'single() can not be used on resultsets prefetching has_many. Use find( \%cond ) or next() instead'
+    );
+  }
+
   if ($where) {
     if (defined $attrs->{where}) {
       $attrs->{where} = {
@@ -545,9 +742,10 @@ sub single {
     $attrs->{where}, $attrs
   );
 
-  return (@data ? ($self->_construct_object(@data))[0] : ());
+  return (@data ? ($self->_construct_object(@data))[0] : undef);
 }
 
+
 # _is_unique_query
 #
 # Try to determine if the specified query is guaranteed to be unique, based on
@@ -592,19 +790,16 @@ sub _collapse_query {
   if (ref $query eq 'ARRAY') {
     foreach my $subquery (@$query) {
       next unless ref $subquery;  # -or
-#      warn "ARRAY: " . Dumper $subquery;
       $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}}) {
-#        warn "HASH: " . Dumper $subquery;
         $collapsed = $self->_collapse_query($subquery, $collapsed);
       }
     }
     else {
-#      warn "LEAF: " . Dumper $query;
       foreach my $col (keys %$query) {
         my $value = $query->{$col};
         $collapsed->{$col}{$value}++;
@@ -651,15 +846,29 @@ sub get_column {
   $cd_rs = $rs->search_like({ title => '%blue%'});
 
 Performs a search, but uses C<LIKE> instead of C<=> as the condition. Note
-that this is simply a convenience method. You most likely want to use
-L</search> with specific operators.
+that this is simply a convenience method retained for ex Class::DBI users.
+You most likely want to use L</search> with specific operators.
 
 For more information, see L<DBIx::Class::Manual::Cookbook>.
 
+This method is deprecated and will be removed in 0.09. Use L</search()>
+instead. An example conversion is:
+
+  ->search_like({ foo => 'bar' });
+
+  # Becomes
+
+  ->search({ foo => { like => 'bar' } });
+
 =cut
 
 sub search_like {
   my $class = shift;
+  carp (
+    'search_like() is deprecated and will be removed in DBIC version 0.09.'
+   .' Instead use ->search({ x => { -like => "y%" } })'
+   .' (note the outer pair of {}s - they are important!)'
+  );
   my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
   my $query = ref $_[0] eq 'HASH' ? { %{shift()} }: {@_};
   $query->{$_} = { 'like' => $query->{$_} } for keys %$query;
@@ -740,7 +949,7 @@ sub next {
       ? @{delete $self->{stashed_row}}
       : $self->cursor->next
   );
-  return unless (@row);
+  return undef unless (@row);
   my ($row, @more) = $self->_construct_object(@row);
   $self->{stashed_objects} = \@more if @more;
   return $row;
@@ -748,7 +957,9 @@ sub next {
 
 sub _construct_object {
   my ($self, @row) = @_;
-  my $info = $self->_collapse_result($self->{_attrs}{as}, \@row);
+
+  my $info = $self->_collapse_result($self->{_attrs}{as}, \@row)
+    or return ();
   my @new = $self->result_class->inflate_result($self->result_source, @$info);
   @new = $self->{_attrs}{record_filter}->(@new)
     if exists $self->{_attrs}{record_filter};
@@ -758,6 +969,19 @@ sub _construct_object {
 sub _collapse_result {
   my ($self, $as_proto, $row) = @_;
 
+  # if the first row that ever came in is totally empty - this means we got
+  # hit by a smooth^Wempty left-joined resultset. Just noop in that case
+  # instead of producing a {}
+  #
+  my $has_def;
+  for (@$row) {
+    if (defined $_) {
+      $has_def++;
+      last;
+    }
+  }
+  return undef unless $has_def;
+
   my @copy = @$row;
 
   # 'foo'         => [ undef, 'foo' ]
@@ -801,7 +1025,7 @@ sub _collapse_result {
   do { # no need to check anything at the front, we always want the first row
 
     my %const;
-  
+
     foreach my $this_as (@construct_as) {
       $const{$this_as->[0]||''}{$this_as->[1]} = shift(@copy);
     }
@@ -835,8 +1059,6 @@ sub _collapse_result {
 
   my @const_keys;
 
-  use Data::Dumper;
-
   foreach my $const (@const_rows) {
     scalar @const_keys or do {
       @const_keys = sort { length($a) <=> length($b) } keys %$const;
@@ -850,7 +1072,7 @@ sub _collapse_result {
         foreach my $p (@parts) {
           $target = $target->[1]->{$p} ||= [];
           $cur .= ".${p}";
-          if ($cur eq ".${key}" && (my @ckey = @{$collapse{$cur}||[]})) { 
+          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) {
@@ -901,12 +1123,25 @@ is derived.
 
 =back
 
-An accessor for the class to use when creating row objects. Defaults to 
-C<< result_source->result_class >> - which in most cases is the name of the 
+An accessor for the class to use when creating row objects. Defaults to
+C<< result_source->result_class >> - which in most cases is the name of the
 L<"table"|DBIx::Class::Manual::Glossary/"ResultSource"> class.
 
+Note that changing the result_class will also remove any components
+that were originally loaded in the source class via
+L<DBIx::Class::ResultSource/load_components>. Any overloaded methods
+in the original source class will not run.
+
 =cut
 
+sub result_class {
+  my ($self, $result_class) = @_;
+  if ($result_class) {
+    $self->ensure_class_loaded($result_class);
+    $self->_result_class($result_class);
+  }
+  $self->_result_class;
+}
 
 =head2 count
 
@@ -919,14 +1154,8 @@ L<"table"|DBIx::Class::Manual::Glossary/"ResultSource"> class.
 =back
 
 Performs an SQL C<COUNT> with the same query as the resultset was built
-with to find the number of elements. If passed arguments, does a search
-on the resultset and counts the results of that.
-
-Note: When using C<count> with C<group_by>, L<DBIX::Class> emulates C<GROUP BY>
-using C<COUNT( DISTINCT( columns ) )>. Some databases (notably SQLite) do
-not support C<DISTINCT> with multiple columns. If you are using such a
-database, you should only use columns from the main table in your C<group_by>
-clause.
+with to find the number of elements. Passing arguments is equivalent to
+C<< $rs->search ($cond, \%attrs)->count >>
 
 =cut
 
@@ -934,47 +1163,206 @@ sub count {
   my $self = shift;
   return $self->search(@_)->count if @_ and defined $_[0];
   return scalar @{ $self->get_cache } if $self->get_cache;
-  my $count = $self->_count;
-  return 0 unless $count;
 
-  $count -= $self->{attrs}{offset} if $self->{attrs}{offset};
-  $count = $self->{attrs}{rows} if
-    $self->{attrs}{rows} and $self->{attrs}{rows} < $count;
+  my $attrs = $self->_resolved_attrs_copy;
+
+  # this is a little optimization - it is faster to do the limit
+  # adjustments in software, instead of a subquery
+  my $rows = delete $attrs->{rows};
+  my $offset = delete $attrs->{offset};
+
+  my $crs;
+  if ($self->_has_resolved_attr (qw/collapse group_by/)) {
+    $crs = $self->_count_subq_rs ($attrs);
+  }
+  else {
+    $crs = $self->_count_rs ($attrs);
+  }
+  my $count = $crs->next;
+
+  $count -= $offset if $offset;
+  $count = $rows if $rows and $rows < $count;
+  $count = 0 if ($count < 0);
+
   return $count;
 }
 
-sub _count { # Separated out so pager can get the full count
+=head2 count_rs
+
+=over 4
+
+=item Arguments: $cond, \%attrs??
+
+=item Return Value: $count_rs
+
+=back
+
+Same as L</count> but returns a L<DBIx::Class::ResultSetColumn> object.
+This can be very handy for subqueries:
+
+  ->search( { amount => $some_rs->count_rs->as_query } )
+
+As with regular resultsets the SQL query will be executed only after
+the resultset is accessed via L</next> or L</all>. That would return
+the same single value obtainable via L</count>.
+
+=cut
+
+sub count_rs {
   my $self = shift;
-  my $select = { count => '*' };
-
-  my $attrs = { %{$self->_resolved_attrs} };
-  if (my $group_by = delete $attrs->{group_by}) {
-    delete $attrs->{having};
-    my @distinct = (ref $group_by ?  @$group_by : ($group_by));
-    # todo: try CONCAT for multi-column pk
-    my @pk = $self->result_source->primary_columns;
-    if (@pk == 1) {
-      my $alias = $attrs->{alias};
-      foreach my $column (@distinct) {
-        if ($column =~ qr/^(?:\Q${alias}.\E)?$pk[0]$/) {
-          @distinct = ($column);
-          last;
-        }
-      }
+  return $self->search(@_)->count_rs if @_;
+
+  # this may look like a lack of abstraction (count() does about the same)
+  # but in fact an _rs *must* use a subquery for the limits, as the
+  # software based limiting can not be ported if this $rs is to be used
+  # in a subquery itself (i.e. ->as_query)
+  if ($self->_has_resolved_attr (qw/collapse group_by offset rows/)) {
+    return $self->_count_subq_rs;
+  }
+  else {
+    return $self->_count_rs;
+  }
+}
+
+#
+# returns a ResultSetColumn object tied to the count query
+#
+sub _count_rs {
+  my ($self, $attrs) = @_;
+
+  my $rsrc = $self->result_source;
+  $attrs ||= $self->_resolved_attrs;
+
+  my $tmp_attrs = { %$attrs };
+
+  # take off any limits, record_filter is cdbi, and no point of ordering a count 
+  delete $tmp_attrs->{$_} for (qw/select as rows offset order_by record_filter/);
+
+  # overwrite the selector (supplied by the storage)
+  $tmp_attrs->{select} = $rsrc->storage->_count_select ($rsrc, $tmp_attrs);
+  $tmp_attrs->{as} = 'count';
+
+  # read the comment on top of the actual function to see what this does
+  $tmp_attrs->{from} = $self->_switch_to_inner_join_if_needed (
+    $tmp_attrs->{from}, $tmp_attrs->{alias}
+  );
+
+  my $tmp_rs = $rsrc->resultset_class->new($rsrc, $tmp_attrs)->get_column ('count');
+
+  return $tmp_rs;
+}
+
+#
+# same as above but uses a subquery
+#
+sub _count_subq_rs {
+  my ($self, $attrs) = @_;
+
+  my $rsrc = $self->result_source;
+  $attrs ||= $self->_resolved_attrs_copy;
+
+  my $sub_attrs = { %$attrs };
+
+  # extra selectors do not go in the subquery and there is no point of ordering it
+  delete $sub_attrs->{$_} for qw/collapse prefetch_select select as order_by/;
+
+  # if we prefetch, we group_by primary keys only as this is what we would get out of the rs via ->next/->all
+  # clobber old group_by regardless
+  if ( keys %{$attrs->{collapse}} ) {
+    $sub_attrs->{group_by} = [ map { "$attrs->{alias}.$_" } ($rsrc->primary_columns) ]
+  }
+
+  $sub_attrs->{select} = $rsrc->storage->_subq_count_select ($rsrc, $sub_attrs);
+
+  # read the comment on top of the actual function to see what this does
+  $sub_attrs->{from} = $self->_switch_to_inner_join_if_needed (
+    $sub_attrs->{from}, $sub_attrs->{alias}
+  );
+
+  $attrs->{from} = [{
+    count_subq => $rsrc->resultset_class->new ($rsrc, $sub_attrs )->as_query
+  }];
+
+  # the subquery replaces this
+  delete $attrs->{$_} for qw/where bind collapse group_by having having_bind rows offset/;
+
+  return $self->_count_rs ($attrs);
+}
+
+
+# The DBIC relationship chaining implementation is pretty simple - every
+# new related_relationship is pushed onto the {from} stack, and the {select}
+# window simply slides further in. This means that when we count somewhere
+# in the middle, we got to make sure that everything in the join chain is an
+# actual inner join, otherwise the count will come back with unpredictable
+# results (a resultset may be generated with _some_ rows regardless of if
+# the relation which the $rs currently selects has rows or not). E.g.
+# $artist_rs->cds->count - normally generates:
+# SELECT COUNT( * ) FROM artist me LEFT JOIN cd cds ON cds.artist = me.artistid
+# which actually returns the number of artists * (number of cds || 1)
+#
+# So what we do here is crawl {from}, determine if the current alias is at
+# the top of the stack, and if not - make sure the chain is inner-joined down
+# to the root.
+#
+sub _switch_to_inner_join_if_needed {
+  my ($self, $from, $alias) = @_;
+
+  return $from if (
+    ref $from ne 'ARRAY'
+      ||
+    ref $from->[0] ne 'HASH'
+      ||
+    ! $from->[0]{-alias}
+      ||
+    $from->[0]{-alias} eq $alias
+  );
+
+  # this would be the case with a subquery - we'll never find
+  # the target as it is not in the parseable part of {from}
+  return $from if @$from == 1;
+
+  my $switch_branch;
+  JOINSCAN:
+  for my $j (@{$from}[1 .. $#$from]) {
+    if ($j->[0]{-alias} eq $alias) {
+      $switch_branch = $j->[0]{-join_path};
+      last JOINSCAN;
     }
+  }
 
-    $select = { count => { distinct => \@distinct } };
+  # something else went wrong
+  return $from unless $switch_branch;
+
+  # So it looks like we will have to switch some stuff around.
+  # local() is useless here as we will be leaving the scope
+  # anyway, and deep cloning is just too fucking expensive
+  # So replace the inner hashref manually
+  my @new_from = ($from->[0]);
+  my $sw_idx = { map { $_ => 1 } @$switch_branch };
+
+  for my $j (@{$from}[1 .. $#$from]) {
+    my $jalias = $j->[0]{-alias};
+
+    if ($sw_idx->{$jalias}) {
+      my %attrs = %{$j->[0]};
+      delete $attrs{-join_type};
+      push @new_from, [
+        \%attrs,
+        @{$j}[ 1 .. $#$j ],
+      ];
+    }
+    else {
+      push @new_from, $j;
+    }
   }
 
-  $attrs->{select} = $select;
-  $attrs->{as} = [qw/count/];
+  return \@new_from;
+}
 
-  # offset, order by and page are not needed to count. record_filter is cdbi
-  delete $attrs->{$_} for qw/rows offset order_by page pager record_filter/;
 
-  my $tmp_rs = (ref $self)->new($self->_source_handle, $attrs);
-  my ($count) = $tmp_rs->cursor->next;
-  return $count;
+sub _bool {
+  return 1;
 }
 
 =head2 count_literal
@@ -1010,18 +1398,21 @@ is returned in list context.
 =cut
 
 sub all {
-  my ($self) = @_;
+  my $self = shift;
+  if(@_) {
+      $self->throw_exception("all() doesn't take any arguments, you probably wanted ->search(...)->all()");
+  }
+
   return @{ $self->get_cache } if $self->get_cache;
 
   my @obj;
 
-  # TODO: don't call resolve here
   if (keys %{$self->_resolved_attrs->{collapse}}) {
-#  if ($self->{attrs}{prefetch}) {
-      # 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
+    # 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));
@@ -1034,6 +1425,7 @@ sub all {
   }
 
   $self->set_cache(\@obj) if $self->{attrs}{cache};
+
   return @obj;
 }
 
@@ -1048,6 +1440,8 @@ sub all {
 =back
 
 Resets the resultset's cursor, so you can iterate through the elements again.
+Implicitly resets the storage cursor, so a subsequent L</next> will trigger
+another query.
 
 =cut
 
@@ -1078,6 +1472,72 @@ sub first {
   return $_[0]->reset->next;
 }
 
+
+# _rs_update_delete
+#
+# Determines whether and what type of subquery is required for the $rs operation.
+# If grouping is necessary either supplies its own, or verifies the current one
+# After all is done delegates to the proper storage method.
+
+sub _rs_update_delete {
+  my ($self, $op, $values) = @_;
+
+  my $rsrc = $self->result_source;
+
+  my $needs_group_by_subq = $self->_has_resolved_attr (qw/collapse group_by -join/);
+  my $needs_subq = $self->_has_resolved_attr (qw/row offset/);
+
+  if ($needs_group_by_subq or $needs_subq) {
+
+    # make a new $rs selecting only the PKs (that's all we really need)
+    my $attrs = $self->_resolved_attrs_copy;
+
+    delete $attrs->{$_} for qw/collapse select as/;
+    $attrs->{columns} = [ map { "$attrs->{alias}.$_" } ($self->result_source->primary_columns) ];
+
+    if ($needs_group_by_subq) {
+      # make sure no group_by was supplied, or if there is one - make sure it matches
+      # the columns compiled above perfectly. Anything else can not be sanely executed
+      # on most databases so croak right then and there
+
+      if (my $g = $attrs->{group_by}) {
+        my @current_group_by = map
+          { $_ =~ /\./ ? $_ : "$attrs->{alias}.$_" }
+          (ref $g eq 'ARRAY' ? @$g : $g );
+
+        if (
+          join ("\x00", sort @current_group_by)
+            ne
+          join ("\x00", sort @{$attrs->{columns}} )
+        ) {
+          $self->throw_exception (
+            "You have just attempted a $op operation on a resultset which does group_by"
+            . ' on columns other than the primary keys, while DBIC internally needs to retrieve'
+            . ' the primary keys in a subselect. All sane RDBMS engines do not support this'
+            . ' kind of queries. Please retry the operation with a modified group_by or'
+            . ' without using one at all.'
+          );
+        }
+      }
+      else {
+        $attrs->{group_by} = $attrs->{columns};
+      }
+    }
+
+    my $subrs = (ref $self)->new($rsrc, $attrs);
+
+    return $self->result_source->storage->_subq_update_delete($subrs, $op, $values);
+  }
+  else {
+    return $rsrc->storage->$op(
+      $rsrc,
+      $op eq 'update' ? $values : (),
+      $self->_cond_for_update_delete,
+    );
+  }
+}
+
+
 # _cond_for_update_delete
 #
 # update/delete require the condition to be modified to handle
@@ -1107,11 +1567,9 @@ sub _cond_for_update_delete {
   elsif (ref $full_cond eq 'HASH') {
     if ((keys %{$full_cond})[0] eq '-and') {
       $cond->{-and} = [];
-
       my @cond = @{$full_cond->{-and}};
-      for (my $i = 0; $i < @cond; $i++) {
+       for (my $i = 0; $i < @cond; $i++) {
         my $entry = $cond[$i];
-
         my $hash;
         if (ref $entry eq 'HASH') {
           $hash = $self->_cond_for_update_delete($entry);
@@ -1120,7 +1578,6 @@ sub _cond_for_update_delete {
           $entry =~ /([^.]+)$/;
           $hash->{$1} = $cond[++$i];
         }
-
         push @{$cond->{-and}}, $hash;
       }
     }
@@ -1132,9 +1589,7 @@ sub _cond_for_update_delete {
     }
   }
   else {
-    $self->throw_exception(
-      "Can't update/delete on resultset with condition unless hash or array"
-    );
+    $self->throw_exception("Can't update/delete on resultset with condition unless hash or array");
   }
 
   return $cond;
@@ -1159,14 +1614,10 @@ if no records were updated; exact type of success value is storage-dependent.
 
 sub update {
   my ($self, $values) = @_;
-  $self->throw_exception("Values for update must be a hash")
+  $self->throw_exception('Values for update must be a hash')
     unless ref $values eq 'HASH';
 
-  my $cond = $self->_cond_for_update_delete;
-   
-  return $self->result_source->storage->update(
-    $self->result_source, $values, $cond
-  );
+  return $self->_rs_update_delete ('update', $values);
 }
 
 =head2 update_all
@@ -1186,7 +1637,7 @@ will run DBIC cascade triggers, while L</update> will not.
 
 sub update_all {
   my ($self, $values) = @_;
-  $self->throw_exception("Values for update must be a hash")
+  $self->throw_exception('Values for update_all must be a hash')
     unless ref $values eq 'HASH';
   foreach my $obj ($self->all) {
     $obj->set_columns($values)->update;
@@ -1200,7 +1651,7 @@ sub update_all {
 
 =item Arguments: none
 
-=item Return Value: 1
+=item Return Value: $storage_rv
 
 =back
 
@@ -1208,15 +1659,17 @@ Deletes the contents of the resultset from its result source. Note that this
 will not run DBIC cascade triggers. See L</delete_all> if you need triggers
 to run. See also L<DBIx::Class::Row/delete>.
 
+Return value will be the amount of rows deleted; exact type of return value
+is storage-dependent.
+
 =cut
 
 sub delete {
-  my ($self) = @_;
-
-  my $cond = $self->_cond_for_update_delete;
+  my $self = shift;
+  $self->throw_exception('delete does not accept any arguments')
+    if @_;
 
-  $self->result_source->storage->delete($self->result_source, $cond);
-  return 1;
+  return $self->_rs_update_delete ('delete');
 }
 
 =head2 delete_all
@@ -1235,7 +1688,10 @@ will run DBIC cascade triggers, while L</delete> will not.
 =cut
 
 sub delete_all {
-  my ($self) = @_;
+  my $self = shift;
+  $self->throw_exception('delete_all does not accept any arguments')
+    if @_;
+
   $_->delete for $self->all;
   return 1;
 }
@@ -1244,27 +1700,29 @@ sub delete_all {
 
 =over 4
 
-=item Arguments: $source_name, \@data;
+=item Arguments: \@data;
 
 =back
 
-Pass an arrayref of hashrefs. Each hashref should be a structure suitable for
-submitting to a $resultset->create(...) method.
+Accepts either an arrayref of hashrefs or alternatively an arrayref of arrayrefs.
+For the arrayref of hashrefs style each hashref should be a structure suitable
+forsubmitting to a $resultset->create(...) method.
 
 In void context, C<insert_bulk> in L<DBIx::Class::Storage::DBI> is used
 to insert the data, as this is a faster method.
 
 Otherwise, each set of data is inserted into the database using
-L<DBIx::Class::ResultSet/create>, and a arrayref of the resulting row
-objects is returned.
+L<DBIx::Class::ResultSet/create>, and the resulting objects are
+accumulated into an array. The array itself, or an array reference
+is returned depending on scalar or list context.
 
 Example:  Assuming an Artist Class that has many CDs Classes relating:
 
   my $Artist_rs = $schema->resultset("Artist");
-  
-  ## Void Context Example 
+
+  ## Void Context Example
   $Artist_rs->populate([
-     { artistid => 4, name => 'Manufactured Crap', cds => [ 
+     { artistid => 4, name => 'Manufactured Crap', cds => [
         { title => 'My First CD', year => 2006 },
         { title => 'Yet More Tweeny-Pop crap', year => 2007 },
       ],
@@ -1276,60 +1734,111 @@ Example:  Assuming an Artist Class that has many CDs Classes relating:
       ],
      },
   ]);
-  
+
   ## Array Context Example
   my ($ArtistOne, $ArtistTwo, $ArtistThree) = $Artist_rs->populate([
     { name => "Artist One"},
-	{ name => "Artist Two"},
-	{ name => "Artist Three", cds=> [
-	  { title => "First CD", year => 2007},
-	  { title => "Second CD", year => 2008},
-	]}
+    { name => "Artist Two"},
+    { name => "Artist Three", cds=> [
+    { title => "First CD", year => 2007},
+    { title => "Second CD", year => 2008},
+  ]}
   ]);
-  
+
   print $ArtistOne->name; ## response is 'Artist One'
   print $ArtistThree->cds->count ## reponse is '2'
 
+For the arrayref of arrayrefs style,  the first element should be a list of the
+fieldsnames to which the remaining elements are rows being inserted.  For
+example:
+
+  $Arstist_rs->populate([
+    [qw/artistid name/],
+    [100, 'A Formally Unknown Singer'],
+    [101, 'A singer that jumped the shark two albums ago'],
+    [102, 'An actually cool singer.'],
+  ]);
+
+Please note an important effect on your data when choosing between void and
+wantarray context. Since void context goes straight to C<insert_bulk> in
+L<DBIx::Class::Storage::DBI> this will skip any component that is overriding
+C<insert>.  So if you are using something like L<DBIx-Class-UUIDColumns> to
+create primary keys for you, you will find that your PKs are empty.  In this
+case you will have to use the wantarray context in order to create those
+values.
+
 =cut
 
 sub populate {
-  my ($self, $data) = @_;
-  
+  my $self = shift @_;
+  my $data = ref $_[0][0] eq 'HASH'
+    ? $_[0] : ref $_[0][0] eq 'ARRAY' ? $self->_normalize_populate_args($_[0]) :
+    $self->throw_exception('Populate expects an arrayref of hashes or arrayref of arrayrefs');
+
   if(defined wantarray) {
     my @created;
     foreach my $item (@$data) {
       push(@created, $self->create($item));
     }
-    return @created;
+    return wantarray ? @created : \@created;
   } else {
     my ($first, @rest) = @$data;
 
-    my @names = grep { !ref $first->{$_} } keys %$first;
+    my @names = grep {!ref $first->{$_}} keys %$first;
+    my @rels = grep { $self->result_source->has_relationship($_) } keys %$first;
+    my @pks = $self->result_source->primary_columns;
+
+    ## do the belongs_to relationships
+    foreach my $index (0..$#$data) {
+
+      # delegate to create() for any dataset without primary keys with specified relationships
+      if (grep { !defined $data->[$index]->{$_} } @pks ) {
+        for my $r (@rels) {
+          if (grep { ref $data->[$index]{$r} eq $_ } qw/HASH ARRAY/) {  # a related set must be a HASH or AoH
+            my @ret = $self->populate($data);
+            return;
+          }
+        }
+      }
+
+      foreach my $rel (@rels) {
+        next unless ref $data->[$index]->{$rel} eq "HASH";
+        my $result = $self->related_resultset($rel)->create($data->[$index]->{$rel});
+        my ($reverse) = keys %{$self->result_source->reverse_relationship_info($rel)};
+        my $related = $result->result_source->_resolve_condition(
+          $result->result_source->relationship_info($reverse)->{cond},
+          $self,
+          $result,
+        );
+
+        delete $data->[$index]->{$rel};
+        $data->[$index] = {%{$data->[$index]}, %$related};
+
+        push @names, keys %$related if $index == 0;
+      }
+    }
 
-    my @values = map {
-      [ map {
-         defined $_ ? $_ : $self->throw_exception("Undefined value for column!")
-      } @$_{@names} ]
-    } @$data;
+    ## do bulk insert on current row
+    my @values = map { [ @$_{@names} ] } @$data;
 
     $self->result_source->storage->insert_bulk(
-      $self->result_source, 
-      \@names, 
+      $self->result_source,
+      \@names,
       \@values,
     );
 
-    my @rels = grep { $self->result_source->has_relationship($_) } keys %$first;
-    my @pks = $self->result_source->primary_columns;
-
+    ## do the has_many relationships
     foreach my $item (@$data) {
 
       foreach my $rel (@rels) {
-        next unless $item->{$rel};
+        next unless $item->{$rel} && ref $item->{$rel} eq "ARRAY";
+
+        my $parent = $self->find(map {{$_=>$item->{$_}} } @pks)
+     || $self->throw_exception('Cannot find the relating object.');
 
-        my $parent = $self->find(map {{$_=>$item->{$_}} } @pks) || next;
         my $child = $parent->$rel;
-		
-        my $related = $child->result_source->resolve_condition(
+
+        my $related = $child->result_source->_resolve_condition(
           $parent->result_source->relationship_info($rel)->{cond},
           $child,
           $parent,
@@ -1344,6 +1853,28 @@ sub populate {
   }
 }
 
+=head2 _normalize_populate_args ($args)
+
+Private method used by L</populate> to normalize its incoming arguments.  Factored
+out in case you want to subclass and accept new argument structures to the
+L</populate> method.
+
+=cut
+
+sub _normalize_populate_args {
+  my ($self, $data) = @_;
+  my @names = @{shift(@$data)};
+  my @results_to_create;
+  foreach my $datum (@$data) {
+    my %result_to_create;
+    foreach my $index (0..$#names) {
+      $result_to_create{$names[$index]} = $$datum[$index];
+    }
+    push @results_to_create, \%result_to_create;
+  }
+  return \@results_to_create;
+}
+
 =head2 pager
 
 =over 4
@@ -1357,19 +1888,35 @@ sub populate {
 Return Value a L<Data::Page> object for the current resultset. Only makes
 sense for queries with a C<page> attribute.
 
+To get the full count of entries for a paged resultset, call
+C<total_entries> on the L<Data::Page> object.
+
 =cut
 
 sub pager {
   my ($self) = @_;
+
+  return $self->{pager} if $self->{pager};
+
   my $attrs = $self->{attrs};
   $self->throw_exception("Can't create pager for non-paged rs")
     unless $self->{attrs}{page};
   $attrs->{rows} ||= 10;
-  return $self->{pager} ||= Data::Page->new(
-    $self->_count, $attrs->{rows}, $self->{attrs}{page});
-}
 
-=head2 page
+  # throw away the paging flags and re-run the count (possibly
+  # with a subselect) to get the real total count
+  my $count_attrs = { %$attrs };
+  delete $count_attrs->{$_} for qw/rows offset page pager/;
+  my $total_count = (ref $self)->new($self->result_source, $count_attrs)->count;
+
+  return $self->{pager} = Data::Page->new(
+    $total_count,
+    $attrs->{rows},
+    $self->{attrs}{page}
+  );
+}
+
+=head2 page
 
 =over 4
 
@@ -1387,7 +1934,7 @@ attribute set on the resultset (10 by default).
 
 sub page {
   my ($self, $page) = @_;
-  return (ref $self)->new($self->_source_handle, { %{$self->{attrs}}, page => $page });
+  return (ref $self)->new($self->result_source, { %{$self->{attrs}}, page => $page });
 }
 
 =head2 new_result
@@ -1396,11 +1943,16 @@ sub page {
 
 =item Arguments: \%vals
 
-=item Return Value: $object
+=item Return Value: $rowobject
 
 =back
 
-Creates an object in the resultset's result class and returns it.
+Creates a new row object in the resultset's result class and returns
+it. The row is not inserted into the database at this point, call
+L<DBIx::Class::Row/insert> to do that. Calling L<DBIx::Class::Row/in_storage>
+will tell you whether the row object has been inserted or not.
+
+Passes the hashref of input on to L<DBIx::Class::Row/new>.
 
 =cut
 
@@ -1408,21 +1960,116 @@ sub new_result {
   my ($self, $values) = @_;
   $self->throw_exception( "new_result needs a hash" )
     unless (ref $values eq 'HASH');
-  $self->throw_exception(
-    "Can't abstract implicit construct, condition not a hash"
-  ) if ($self->{cond} && !(ref $self->{cond} eq 'HASH'));
 
+  my %new;
   my $alias = $self->{attrs}{alias};
-  my $collapsed_cond = $self->{cond} ? $self->_collapse_cond($self->{cond}) : {};
-  my %new = (
+
+  if (
+    defined $self->{cond}
+    && $self->{cond} eq $DBIx::Class::ResultSource::UNRESOLVABLE_CONDITION
+  ) {
+    %new = %{ $self->{attrs}{related_objects} || {} };  # nothing might have been inserted yet
+    $new{-from_resultset} = [ keys %new ] if keys %new;
+  } else {
+    $self->throw_exception(
+      "Can't abstract implicit construct, condition not a hash"
+    ) if ($self->{cond} && !(ref $self->{cond} eq 'HASH'));
+
+    my $collapsed_cond = (
+      $self->{cond}
+        ? $self->_collapse_cond($self->{cond})
+        : {}
+    );
+
+    # precendence must be given to passed values over values inherited from
+    # the cond, so the order here is important.
+    my %implied =  %{$self->_remove_alias($collapsed_cond, $alias)};
+    while( my($col,$value) = each %implied ){
+      if(ref($value) eq 'HASH' && keys(%$value) && (keys %$value)[0] eq '='){
+        $new{$col} = $value->{'='};
+        next;
+      }
+      $new{$col} = $value if $self->_is_deterministic_value($value);
+    }
+  }
+
+  %new = (
+    %new,
     %{ $self->_remove_alias($values, $alias) },
-    %{ $self->_remove_alias($collapsed_cond, $alias) },
-    -source_handle => $self->_source_handle
+    -source_handle => $self->_source_handle,
+    -result_source => $self->result_source, # DO NOT REMOVE THIS, REQUIRED
   );
 
   return $self->result_class->new(\%new);
 }
 
+# _is_deterministic_value
+#
+# Make an effor to strip non-deterministic values from the condition,
+# to make sure new_result chokes less
+
+sub _is_deterministic_value {
+  my $self = shift;
+  my $value = shift;
+  my $ref_type = ref $value;
+  return 1 if $ref_type eq '' || $ref_type eq 'SCALAR';
+  return 1 if Scalar::Util::blessed($value);
+  return 0;
+}
+
+# _has_resolved_attr
+#
+# determines if the resultset defines at least one
+# of the attributes supplied
+#
+# used to determine if a subquery is neccessary
+#
+# supports some virtual attributes:
+#   -join
+#     This will scan for any joins being present on the resultset.
+#     It is not a mere key-search but a deep inspection of {from}
+#
+
+sub _has_resolved_attr {
+  my ($self, @attr_names) = @_;
+
+  my $attrs = $self->_resolved_attrs;
+
+  my %extra_checks;
+
+  for my $n (@attr_names) {
+    if (grep { $n eq $_ } (qw/-join/) ) {
+      $extra_checks{$n}++;
+      next;
+    }
+
+    my $attr =  $attrs->{$n};
+
+    next if not defined $attr;
+
+    if (ref $attr eq 'HASH') {
+      return 1 if keys %$attr;
+    }
+    elsif (ref $attr eq 'ARRAY') {
+      return 1 if @$attr;
+    }
+    else {
+      return 1 if $attr;
+    }
+  }
+
+  # a resolved join is expressed as a multi-level from
+  return 1 if (
+    $extra_checks{-join}
+      and
+    ref $attrs->{from} eq 'ARRAY'
+      and
+    @{$attrs->{from}} > 1
+  );
+
+  return 0;
+}
+
 # _collapse_cond
 #
 # Recursively collapse the condition.
@@ -1435,19 +2082,16 @@ sub _collapse_cond {
   if (ref $cond eq 'ARRAY') {
     foreach my $subcond (@$cond) {
       next unless ref $subcond;  # -or
-#      warn "ARRAY: " . Dumper $subcond;
       $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}}) {
-#        warn "HASH: " . Dumper $subcond;
         $collapsed = $self->_collapse_cond($subcond, $collapsed);
       }
     }
     else {
-#      warn "LEAF: " . Dumper $cond;
       foreach my $col (keys %$cond) {
         my $value = $cond->{$col};
         $collapsed->{$col} = $value;
@@ -1481,30 +2125,84 @@ sub _remove_alias {
   return \%unaliased;
 }
 
+=head2 as_query (EXPERIMENTAL)
+
+=over 4
+
+=item Arguments: none
+
+=item Return Value: \[ $sql, @bind ]
+
+=back
+
+Returns the SQL query and bind vars associated with the invocant.
+
+This is generally used as the RHS for a subquery.
+
+B<NOTE>: This feature is still experimental.
+
+=cut
+
+sub as_query {
+  my $self = shift;
+
+  my $attrs = $self->_resolved_attrs_copy;
+
+  # For future use:
+  #
+  # in list ctx:
+  # my ($sql, \@bind, \%dbi_bind_attrs) = _select_args_to_query (...)
+  # $sql also has no wrapping parenthesis in list ctx
+  #
+  my $sqlbind = $self->result_source->storage
+    ->_select_args_to_query ($attrs->{from}, $attrs->{select}, $attrs->{where}, $attrs);
+
+  return $sqlbind;
+}
+
 =head2 find_or_new
 
 =over 4
 
 =item Arguments: \%vals, \%attrs?
 
-=item Return Value: $object
+=item Return Value: $rowobject
 
 =back
 
-Find an existing record from this resultset. If none exists, instantiate a new
-result object and return it. The object will not be saved into your storage
+  my $artist = $schema->resultset('Artist')->find_or_new(
+    { artist => 'fred' }, { key => 'artists' });
+
+  $cd->cd_to_producer->find_or_new({ producer => $producer },
+                                   { key => 'primary });
+
+Find an existing record from this resultset, based on its primary
+key, or a unique constraint. If none exists, instantiate a new result
+object and return it. The object will not be saved into your storage
 until you call L<DBIx::Class::Row/insert> on it.
 
+You most likely want this method when looking for existing rows using
+a unique constraint that is not the primary key, or looking for
+related rows.
+
 If you want objects to be saved immediately, use L</find_or_create> instead.
 
+B<Note>: C<find_or_new> is probably not what you want when creating a
+new row in a table that uses primary keys supplied by the
+database. Passing in a primary key column with a value of I<undef>
+will cause L</find> to attempt to search for a row with a value of
+I<NULL>.
+
 =cut
 
 sub find_or_new {
   my $self     = shift;
   my $attrs    = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
   my $hash     = ref $_[0] eq 'HASH' ? shift : {@_};
-  my $exists   = $self->find($hash, $attrs);
-  return defined $exists ? $exists : $self->new_result($hash);
+  if (keys %$hash and my $row = $self->find($hash, $attrs) ) {
+    return $row;
+  }
+  return $self->new_result($hash);
 }
 
 =head2 create
@@ -1513,14 +2211,66 @@ sub find_or_new {
 
 =item Arguments: \%vals
 
-=item Return Value: $object
+=item Return Value: a L<DBIx::Class::Row> $object
 
 =back
 
-Inserts a record into the resultset and returns the object representing it.
+Attempt to create a single new row or a row with multiple related rows
+in the table represented by the resultset (and related tables). This
+will not check for duplicate rows before inserting, use
+L</find_or_create> to do that.
+
+To create one row for this resultset, pass a hashref of key/value
+pairs representing the columns of the table and the values you wish to
+store. If the appropriate relationships are set up, foreign key fields
+can also be passed an object representing the foreign row, and the
+value will be set to its primary key.
+
+To create related objects, pass a hashref of related-object column values
+B<keyed on the relationship name>. If the relationship is of type C<multi>
+(L<DBIx::Class::Relationship/has_many>) - pass an arrayref of hashrefs.
+The process will correctly identify columns holding foreign keys, and will
+transparrently populate them from the keys of the corresponding relation.
+This can be applied recursively, and will work correctly for a structure
+with an arbitrary depth and width, as long as the relationships actually
+exists and the correct column data has been supplied.
+
+
+Instead of hashrefs of plain related data (key/value pairs), you may
+also pass new or inserted objects. New objects (not inserted yet, see
+L</new>), will be inserted into their appropriate tables.
 
 Effectively a shortcut for C<< ->new_result(\%vals)->insert >>.
 
+Example of creating a new row.
+
+  $person_rs->create({
+    name=>"Some Person",
+    email=>"somebody@someplace.com"
+  });
+
+Example of creating a new row and also creating rows in a related C<has_many>
+or C<has_one> resultset.  Note Arrayref.
+
+  $artist_rs->create(
+     { artistid => 4, name => 'Manufactured Crap', cds => [
+        { title => 'My First CD', year => 2006 },
+        { title => 'Yet More Tweeny-Pop crap', year => 2007 },
+      ],
+     },
+  );
+
+Example of creating a new row and also creating a row in a related
+C<belongs_to>resultset. Note Hashref.
+
+  $cd_rs->create({
+    title=>"Music for Silly Walks",
+    year=>2000,
+    artist => {
+      name=>"Silly Musician",
+    }
+  });
+
 =cut
 
 sub create {
@@ -1536,13 +2286,14 @@ sub create {
 
 =item Arguments: \%vals, \%attrs?
 
-=item Return Value: $object
+=item Return Value: $rowobject
 
 =back
 
-  $class->find_or_create({ key => $val, ... });
+  $cd->cd_to_producer->find_or_create({ producer => $producer },
+                                      { key => 'primary' });
 
-Tries to find a record based on its primary key or unique constraint; if none
+Tries to find a record based on its primary key or unique constraints; if none
 is found, creates one and returns that instead.
 
   my $cd = $schema->resultset('CD')->find_or_create({
@@ -1563,6 +2314,18 @@ constraint. For example:
     { key => 'cd_artist_title' }
   );
 
+B<Note>: Because find_or_create() reads from the database and then
+possibly inserts based on the result, this method is subject to a race
+condition. Another process could create a record in the table after
+the find has completed and before the create has started. To avoid
+this problem, use find_or_create() inside a transaction.
+
+B<Note>: C<find_or_create> is probably not what you want when creating
+a new row in a table that uses primary keys supplied by the
+database. Passing in a primary key column with a value of I<undef>
+will cause L</find> to attempt to search for a row with a value of
+I<NULL>.
+
 See also L</find> and L</update_or_create>. For information on how to declare
 unique constraints, see L<DBIx::Class::ResultSource/add_unique_constraint>.
 
@@ -1572,8 +2335,10 @@ sub find_or_create {
   my $self     = shift;
   my $attrs    = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
   my $hash     = ref $_[0] eq 'HASH' ? shift : {@_};
-  my $exists   = $self->find($hash, $attrs);
-  return defined $exists ? $exists : $self->create($hash);
+  if (keys %$hash and my $row = $self->find($hash, $attrs) ) {
+    return $row;
+  }
+  return $self->create($hash);
 }
 
 =head2 update_or_create
@@ -1582,11 +2347,11 @@ sub find_or_create {
 
 =item Arguments: \%col_values, { key => $unique_constraint }?
 
-=item Return Value: $object
+=item Return Value: $rowobject
 
 =back
 
-  $class->update_or_create({ col => $val, ... });
+  $resultset->update_or_create({ col => $val, ... });
 
 First, searches for an existing row matching one of the unique constraints
 (including the primary key) on the source of this resultset. If a row is
@@ -1606,6 +2371,14 @@ For example:
     { key => 'cd_artist_title' }
   );
 
+  $cd->cd_to_producer->update_or_create({
+    producer => $producer,
+    name => 'harry',
+  }, {
+    key => 'primary,
+  });
+
+
 If no C<key> is specified, it searches on all unique constraints defined on the
 source, including the primary key.
 
@@ -1614,6 +2387,12 @@ If the C<key> is specified as C<primary>, it searches only on the primary key.
 See also L</find> and L</find_or_create>. For information on how to declare
 unique constraints, see L<DBIx::Class::ResultSource/add_unique_constraint>.
 
+B<Note>: C<update_or_create> is probably not what you want when
+looking for a row in a table that uses primary keys supplied by the
+database, unless you actually have a key value. Passing in a primary
+key column with a value of I<undef> will cause L</find> to attempt to
+search for a row with a value of I<NULL>.
+
 =cut
 
 sub update_or_create {
@@ -1630,6 +2409,63 @@ sub update_or_create {
   return $self->create($cond);
 }
 
+=head2 update_or_new
+
+=over 4
+
+=item Arguments: \%col_values, { key => $unique_constraint }?
+
+=item Return Value: $rowobject
+
+=back
+
+  $resultset->update_or_new({ col => $val, ... });
+
+First, searches for an existing row matching one of the unique constraints
+(including the primary key) on the source of this resultset. If a row is
+found, updates it with the other given column values. Otherwise, instantiate
+a new result object and return it. The object will not be saved into your storage
+until you call L<DBIx::Class::Row/insert> on it.
+
+Takes an optional C<key> attribute to search on a specific unique constraint.
+For example:
+
+  # In your application
+  my $cd = $schema->resultset('CD')->update_or_new(
+    {
+      artist => 'Massive Attack',
+      title  => 'Mezzanine',
+      year   => 1998,
+    },
+    { key => 'cd_artist_title' }
+  );
+
+  if ($cd->in_storage) {
+      # the cd was updated
+  }
+  else {
+      # the cd is not yet in the database, let's insert it
+      $cd->insert;
+  }
+
+See also L</find>, L</find_or_create> and L<find_or_new>.
+
+=cut
+
+sub update_or_new {
+    my $self  = shift;
+    my $attrs = ( @_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {} );
+    my $cond  = ref $_[0] eq 'HASH' ? shift : {@_};
+
+    my $row = $self->find( $cond, $attrs );
+    if ( defined $row ) {
+        $row->update($cond);
+        return $row;
+    }
+
+    return $self->new_result($cond);
+}
+
 =head2 get_cache
 
 =over 4
@@ -1642,6 +2478,9 @@ sub update_or_create {
 
 Gets the contents of the cache for the resultset, if the cache is set.
 
+The cache is populated either by using the L</prefetch> attribute to
+L</search> or by calling L</set_cache>.
+
 =cut
 
 sub get_cache {
@@ -1663,6 +2502,9 @@ of objects of the same class as those produced by the resultset. Note that
 if the cache is set the resultset will return the cached objects rather
 than re-querying the database even if the cache attr is not set.
 
+The contents of the cache can also be populated by using the
+L</prefetch> attribute to L</search>.
+
 =cut
 
 sub set_cache {
@@ -1711,189 +2553,498 @@ sub related_resultset {
 
   $self->{related_resultsets} ||= {};
   return $self->{related_resultsets}{$rel} ||= do {
-    my $rel_obj = $self->result_source->relationship_info($rel);
+    my $rel_info = $self->result_source->relationship_info($rel);
 
     $self->throw_exception(
-      "search_related: result source '" . $self->_source_handle->source_moniker .
+      "search_related: result source '" . $self->result_source->source_name .
         "' has no such relationship $rel")
-      unless $rel_obj;
-    
-    my ($from,$seen) = $self->_resolve_from($rel);
+      unless $rel_info;
+
+    my ($from,$seen) = $self->_chain_relationship($rel);
 
     my $join_count = $seen->{$rel};
     my $alias = ($join_count > 1 ? join('_', $rel, $join_count) : $rel);
 
-    $self->_source_handle->schema->resultset($rel_obj->{class})->search_rs(
-      undef, {
-        %{$self->{attrs}||{}},
-        join => undef,
-        prefetch => undef,
-        select => undef,
-        as => undef,
-        alias => $alias,
-        where => $self->{cond},
-        seen_join => $seen,
-        from => $from,
-    });
+    #XXX - temp fix for result_class bug. There likely is a more elegant fix -groditi
+    my %attrs = %{$self->{attrs}||{}};
+    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 = $self->result_source->related_source($rel);
+
+    my $new = do {
+
+      # The reason we do this now instead of passing the alias to the
+      # search_rs below is that if you wrap/overload resultset on the
+      # source you need to know what alias it's -going- to have for things
+      # to work sanely (e.g. RestrictWithObject wants to be able to add
+      # extra query restrictions, and these may need to be $alias.)
+
+      my $attrs = $rel_source->resultset_attributes;
+      local $attrs->{alias} = $alias;
+
+      $rel_source->resultset
+                 ->search_rs(
+                     undef, {
+                       %attrs,
+                       join => undef,
+                       prefetch => undef,
+                       select => undef,
+                       as => undef,
+                       where => $self->{cond},
+                       seen_join => $seen,
+                       from => $from,
+                   });
+    };
+    $new->set_cache($new_cache) if $new_cache;
+    $new;
   };
 }
 
-sub _resolve_from {
-  my ($self, $extra_join) = @_;
+=head2 current_source_alias
+
+=over 4
+
+=item Arguments: none
+
+=item Return Value: $source_alias
+
+=back
+
+Returns the current table alias for the result source this resultset is built
+on, that will be used in the SQL query. Usually it is C<me>.
+
+Currently the source alias that refers to the result set returned by a
+L</search>/L</find> family method depends on how you got to the resultset: it's
+C<me> by default, but eg. L</search_related> aliases it to the related result
+source name (and keeps C<me> referring to the original result set). The long
+term goal is to make L<DBIx::Class> always alias the current resultset as C<me>
+(and make this method unnecessary).
+
+Thus it's currently necessary to use this method in predefined queries (see
+L<DBIx::Class::Manual::Cookbook/Predefined searches>) when referring to the
+source alias of the current result set:
+
+  # in a result set class
+  sub modified_by {
+    my ($self, $user) = @_;
+
+    my $me = $self->current_source_alias;
+
+    return $self->search(
+      "$me.modified" => $user->id,
+    );
+  }
+
+=cut
+
+sub current_source_alias {
+  my ($self) = @_;
+
+  return ($self->{attrs} || {})->{alias} || 'me';
+}
+
+# This code is called by search_related, and makes sure there
+# is clear separation between the joins before, during, and
+# after the relationship. This information is needed later
+# in order to properly resolve prefetch aliases (any alias
+# with a relation_chain_depth less than the depth of the
+# current prefetch is not considered)
+#
+# The increments happen in 1/2s to make it easier to correlate the
+# join depth with the join path. An integer means a relationship
+# specified via a search_related, whereas a fraction means an added
+# join/prefetch via attributes
+sub _chain_relationship {
+  my ($self, $rel) = @_;
   my $source = $self->result_source;
   my $attrs = $self->{attrs};
-  
-  my $from = $attrs->{from}
-    || [ { $attrs->{alias} => $source->from } ];
-    
-  my $seen = { %{$attrs->{seen_join}||{}} };
-
-  my $join = ($attrs->{join}
-               ? [ $attrs->{join}, $extra_join ]
-               : $extra_join);
-  $from = [
-    @$from,
-    ($join ? $source->resolve_join($join, $attrs->{alias}, $seen) : ()),
-  ];
+
+  my $from = [ @{
+      $attrs->{from}
+        ||
+      [{
+        -source_handle => $source->handle,
+        -alias => $attrs->{alias},
+        $attrs->{alias} => $source->from,
+      }]
+  }];
+
+  my $seen = { %{$attrs->{seen_join} || {} } };
+  my $jpath = ($attrs->{seen_join} && keys %{$attrs->{seen_join}}) 
+    ? $from->[-1][0]{-join_path} 
+    : [];
+
+
+  # we need to take the prefetch the attrs into account before we
+  # ->_resolve_join as otherwise they get lost - captainL
+  my $merged = $self->_merge_attr( $attrs->{join}, $attrs->{prefetch} );
+
+  my @requested_joins = $source->_resolve_join(
+    $merged,
+    $attrs->{alias},
+    $seen,
+    $jpath,
+  );
+
+  push @$from, @requested_joins;
+
+  $seen->{-relation_chain_depth} += 0.5;
+
+  # if $self already had a join/prefetch specified on it, the requested
+  # $rel might very well be already included. What we do in this case
+  # is effectively a no-op (except that we bump up the chain_depth on
+  # the join in question so we could tell it *is* the search_related)
+  my $already_joined;
+
+
+  # we consider the last one thus reverse
+  for my $j (reverse @requested_joins) {
+    if ($rel eq $j->[0]{-join_path}[-1]) {
+      $j->[0]{-relation_chain_depth} += 0.5;
+      $already_joined++;
+      last;
+    }
+  }
+
+# alternative way to scan the entire chain - not backwards compatible
+#  for my $j (reverse @$from) {
+#    next unless ref $j eq 'ARRAY';
+#    if ($j->[0]{-join_path} && $j->[0]{-join_path}[-1] eq $rel) {
+#      $j->[0]{-relation_chain_depth} += 0.5;
+#      $already_joined++;
+#      last;
+#    }
+#  }
+
+  unless ($already_joined) {
+    push @$from, $source->_resolve_join(
+      $rel,
+      $attrs->{alias},
+      $seen,
+      $jpath,
+    );
+  }
+
+  $seen->{-relation_chain_depth} += 0.5;
 
   return ($from,$seen);
 }
 
+# too many times we have to do $attrs = { %{$self->_resolved_attrs} }
+sub _resolved_attrs_copy {
+  my $self = shift;
+  return { %{$self->_resolved_attrs (@_)} };
+}
+
 sub _resolved_attrs {
   my $self = shift;
   return $self->{_attrs} if $self->{_attrs};
 
-  my $attrs = { %{$self->{attrs}||{}} };
+  my $attrs  = { %{ $self->{attrs} || {} } };
   my $source = $self->result_source;
-  my $alias = $attrs->{alias};
+  my $alias  = $attrs->{alias};
 
   $attrs->{columns} ||= delete $attrs->{cols} if exists $attrs->{cols};
-  if ($attrs->{columns}) {
-    delete $attrs->{as};
-  } elsif (!$attrs->{select}) {
-    $attrs->{columns} = [ $source->columns ];
-  }
- 
-  $attrs->{select} = 
-    ($attrs->{select}
-      ? (ref $attrs->{select} eq 'ARRAY'
-          ? [ @{$attrs->{select}} ]
-          : [ $attrs->{select} ])
-      : [ map { m/\./ ? $_ : "${alias}.$_" } @{delete $attrs->{columns}} ]
-    );
-  $attrs->{as} =
-    ($attrs->{as}
-      ? (ref $attrs->{as} eq 'ARRAY'
-          ? [ @{$attrs->{as}} ]
-          : [ $attrs->{as} ])
-      : [ map { m/^\Q${alias}.\E(.+)$/ ? $1 : $_ } @{$attrs->{select}} ]
+  my @colbits;
+
+  # build columns (as long as select isn't set) into a set of as/select hashes
+  unless ( $attrs->{select} ) {
+      @colbits = map {
+          ( ref($_) eq 'HASH' )
+              ? $_
+              : {
+                  (
+                    /^\Q${alias}.\E(.+)$/
+                      ? "$1"
+                      : "$_"
+                  )
+                =>
+                  (
+                    /\./
+                      ? "$_"
+                      : "${alias}.$_"
+                  )
+            }
+      } ( ref($attrs->{columns}) eq 'ARRAY' ) ? @{ delete $attrs->{columns}} : (delete $attrs->{columns} || $source->columns );
+  }
+  # add the additional columns on
+  foreach ( 'include_columns', '+columns' ) {
+      push @colbits, map {
+          ( ref($_) eq 'HASH' )
+            ? $_
+            : { ( split( /\./, $_ ) )[-1] => ( /\./ ? $_ : "${alias}.$_" ) }
+      } ( ref($attrs->{$_}) eq 'ARRAY' ) ? @{ delete $attrs->{$_} } : delete $attrs->{$_} if ( $attrs->{$_} );
+  }
+
+  # start with initial select items
+  if ( $attrs->{select} ) {
+    $attrs->{select} =
+        ( ref $attrs->{select} eq 'ARRAY' )
+      ? [ @{ $attrs->{select} } ]
+      : [ $attrs->{select} ];
+    $attrs->{as} = (
+      $attrs->{as}
+      ? (
+        ref $attrs->{as} eq 'ARRAY'
+        ? [ @{ $attrs->{as} } ]
+        : [ $attrs->{as} ]
+        )
+      : [ map { m/^\Q${alias}.\E(.+)$/ ? $1 : $_ } @{ $attrs->{select} } ]
     );
-  
-  my $adds;
-  if ($adds = delete $attrs->{include_columns}) {
-    $adds = [$adds] unless ref $adds eq 'ARRAY';
-    push(@{$attrs->{select}}, @$adds);
-    push(@{$attrs->{as}}, map { m/([^.]+)$/; $1 } @$adds);
   }
-  if ($adds = delete $attrs->{'+select'}) {
+  else {
+
+    # otherwise we intialise select & as to empty
+    $attrs->{select} = [];
+    $attrs->{as}     = [];
+  }
+
+  # now add colbits to select/as
+  push( @{ $attrs->{select} }, map { values( %{$_} ) } @colbits );
+  push( @{ $attrs->{as} },     map { keys( %{$_} ) } @colbits );
+
+  my $adds;
+  if ( $adds = delete $attrs->{'+select'} ) {
     $adds = [$adds] unless ref $adds eq 'ARRAY';
-    push(@{$attrs->{select}},
-           map { /\./ || ref $_ ? $_ : "${alias}.$_" } @$adds);
+    push(
+      @{ $attrs->{select} },
+      map { /\./ || ref $_ ? $_ : "${alias}.$_" } @$adds
+    );
   }
-  if (my $adds = delete $attrs->{'+as'}) {
+  if ( $adds = delete $attrs->{'+as'} ) {
     $adds = [$adds] unless ref $adds eq 'ARRAY';
-    push(@{$attrs->{as}}, @$adds);
+    push( @{ $attrs->{as} }, @$adds );
   }
 
-  $attrs->{from} ||= [ { 'me' => $source->from } ];
+  $attrs->{from} ||= [ {
+    -source_handle => $source->handle,
+    -alias => $self->{attrs}{alias},
+    $self->{attrs}{alias} => $source->from,
+  } ];
+
+  if ( $attrs->{join} || $attrs->{prefetch} ) {
+
+    $self->throw_exception ('join/prefetch can not be used with a literal scalarref {from}')
+      if ref $attrs->{from} ne 'ARRAY';
 
-  if (exists $attrs->{join} || exists $attrs->{prefetch}) {
     my $join = delete $attrs->{join} || {};
 
-    if (defined $attrs->{prefetch}) {
-      $join = $self->_merge_attr(
-        $join, $attrs->{prefetch}
-      );
+    if ( defined $attrs->{prefetch} ) {
+      $join = $self->_merge_attr( $join, $attrs->{prefetch} );
     }
 
-    $attrs->{from} =   # have to copy here to avoid corrupting the original
+    $attrs->{from} =    # have to copy here to avoid corrupting the original
       [
-        @{$attrs->{from}}, 
-        $source->resolve_join($join, $alias, { %{$attrs->{seen_join}||{}} })
+        @{ $attrs->{from} },
+        $source->_resolve_join(
+          $join,
+          $alias,
+          { %{ $attrs->{seen_join} || {} } },
+          ($attrs->{seen_join} && keys %{$attrs->{seen_join}})
+            ? $attrs->{from}[-1][0]{-join_path}
+            : []
+          ,
+        )
       ];
   }
 
-  $attrs->{group_by} ||= $attrs->{select} if delete $attrs->{distinct};
-  if ($attrs->{order_by}) {
-    $attrs->{order_by} = (ref($attrs->{order_by}) eq 'ARRAY'
-                           ? [ @{$attrs->{order_by}} ]
-                           : [ $attrs->{order_by} ]);
+  if ( $attrs->{order_by} ) {
+    $attrs->{order_by} = (
+      ref( $attrs->{order_by} ) eq 'ARRAY'
+      ? [ @{ $attrs->{order_by} } ]
+      : [ $attrs->{order_by} ]
+    );
+  }
+
+  if ($attrs->{group_by} and ! ref $attrs->{group_by}) {
+    $attrs->{group_by} = [ $attrs->{group_by} ];
+  }
+
+  # If the order_by is otherwise empty - we will use this for TOP limit
+  # emulation and the like.
+  # Although this is needed only if the order_by is not defined, it is
+  # actually cheaper to just populate this rather than properly examining
+  # order_by (stuf like [ {} ] and the like)
+  $attrs->{_virtual_order_by} = [ $self->result_source->primary_columns ];
+
+
+  $attrs->{collapse} ||= {};
+  if ( my $prefetch = delete $attrs->{prefetch} ) {
+    $prefetch = $self->_merge_attr( {}, $prefetch );
+
+    my $prefetch_ordering = [];
+
+    my $join_map = $self->_joinpath_aliases ($attrs->{from}, $attrs->{seen_join});
+
+    my @prefetch =
+      $source->_resolve_prefetch( $prefetch, $alias, $join_map, $prefetch_ordering, $attrs->{collapse} );
+
+    $attrs->{prefetch_select} = [ map { $_->[0] } @prefetch ];
+    push @{ $attrs->{select} }, @{$attrs->{prefetch_select}};
+    push @{ $attrs->{as} }, (map { $_->[1] } @prefetch);
+
+    push( @{ $attrs->{order_by} }, @$prefetch_ordering );
+    $attrs->{_collapse_order_by} = \@$prefetch_ordering;
+  }
+
+
+  if (delete $attrs->{distinct}) {
+    $attrs->{group_by} ||= [ grep { !ref($_) || (ref($_) ne 'HASH') } @{$attrs->{select}} ];
+  }
+
+  # 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
+  if (my $page = delete $attrs->{page}) {
+    $attrs->{offset} = 
+      ($attrs->{rows} * ($page - 1))
+            +
+      ($attrs->{offset} || 0)
+    ;
+  }
+
+  return $self->{_attrs} = $attrs;
+}
+
+sub _joinpath_aliases {
+  my ($self, $fromspec, $seen) = @_;
+
+  my $paths = {};
+  return $paths unless ref $fromspec eq 'ARRAY';
+
+  my $cur_depth = $seen->{-relation_chain_depth} || 0;
+
+  if (int ($cur_depth) != $cur_depth) {
+    $self->throw_exception ("-relation_chain_depth is not an integer, something went horribly wrong ($cur_depth)");
+  }
+
+  for my $j (@$fromspec) {
+
+    next if ref $j ne 'ARRAY';
+    next if $j->[0]{-relation_chain_depth} < $cur_depth;
+
+    my $jpath = $j->[0]{-join_path};
+
+    my $p = $paths;
+    $p = $p->{$_} ||= {} for @{$jpath}[$cur_depth .. $#$jpath];
+    push @{$p->{-join_aliases} }, $j->[0]{-alias};
+  }
+
+  return $paths;
+}
+
+sub _rollout_attr {
+  my ($self, $attr) = @_;
+
+  if (ref $attr eq 'HASH') {
+    return $self->_rollout_hash($attr);
+  } elsif (ref $attr eq 'ARRAY') {
+    return $self->_rollout_array($attr);
   } else {
-    $attrs->{order_by} = [];    
-  }
-
-  my $collapse = $attrs->{collapse} || {};
-  if (my $prefetch = delete $attrs->{prefetch}) {
-    $prefetch = $self->_merge_attr({}, $prefetch);
-    my @pre_order;
-    my $seen = $attrs->{seen_join} || {};
-    foreach my $p (ref $prefetch eq 'ARRAY' ? @$prefetch : ($prefetch)) {
-      # bring joins back to level of current class
-      my @prefetch = $source->resolve_prefetch(
-        $p, $alias, $seen, \@pre_order, $collapse
-      );
-      push(@{$attrs->{select}}, map { $_->[0] } @prefetch);
-      push(@{$attrs->{as}}, map { $_->[1] } @prefetch);
+    return [$attr];
+  }
+}
+
+sub _rollout_array {
+  my ($self, $attr) = @_;
+
+  my @rolled_array;
+  foreach my $element (@{$attr}) {
+    if (ref $element eq 'HASH') {
+      push( @rolled_array, @{ $self->_rollout_hash( $element ) } );
+    } elsif (ref $element eq 'ARRAY') {
+      #  XXX - should probably recurse here
+      push( @rolled_array, @{$self->_rollout_array($element)} );
+    } else {
+      push( @rolled_array, $element );
     }
-    push(@{$attrs->{order_by}}, @pre_order);
   }
-  $attrs->{collapse} = $collapse;
+  return \@rolled_array;
+}
 
-  return $self->{_attrs} = $attrs;
+sub _rollout_hash {
+  my ($self, $attr) = @_;
+
+  my @rolled_array;
+  foreach my $key (keys %{$attr}) {
+    push( @rolled_array, { $key => $attr->{$key} } );
+  }
+  return \@rolled_array;
 }
 
-sub _merge_attr {
+sub _calculate_score {
   my ($self, $a, $b) = @_;
-  return $b unless defined($a);
-  return $a unless defined($b);
-  
-  if (ref $b eq 'HASH' && ref $a eq 'HASH') {
-    foreach my $key (keys %{$b}) {
-      if (exists $a->{$key}) {
-        $a->{$key} = $self->_merge_attr($a->{$key}, $b->{$key});
+
+  if (ref $b eq 'HASH') {
+    my ($b_key) = keys %{$b};
+    if (ref $a eq 'HASH') {
+      my ($a_key) = keys %{$a};
+      if ($a_key eq $b_key) {
+        return (1 + $self->_calculate_score( $a->{$a_key}, $b->{$b_key} ));
       } else {
-        $a->{$key} = $b->{$key};
+        return 0;
       }
+    } else {
+      return ($a eq $b_key) ? 1 : 0;
     }
-    return $a;
   } else {
-    $a = [$a] unless ref $a eq 'ARRAY';
-    $b = [$b] unless ref $b eq 'ARRAY';
-
-    my $hash = {};
-    my @array;
-    foreach my $x ($a, $b) {
-      foreach my $element (@{$x}) {
-        if (ref $element eq 'HASH') {
-          $hash = $self->_merge_attr($hash, $element);
-        } elsif (ref $element eq 'ARRAY') {
-          push(@array, @{$element});
-        } else {
-          push(@array, $element) unless $b == $x
-            && grep { $_ eq $element } @array;
-        }
+    if (ref $a eq 'HASH') {
+      my ($a_key) = keys %{$a};
+      return ($b eq $a_key) ? 1 : 0;
+    } else {
+      return ($b eq $a) ? 1 : 0;
+    }
+  }
+}
+
+sub _merge_attr {
+  my ($self, $orig, $import) = @_;
+
+  return $import unless defined($orig);
+  return $orig unless defined($import);
+
+  $orig = $self->_rollout_attr($orig);
+  $import = $self->_rollout_attr($import);
+
+  my $seen_keys;
+  foreach my $import_element ( @{$import} ) {
+    # find best candidate from $orig to merge $b_element into
+    my $best_candidate = { position => undef, score => 0 }; my $position = 0;
+    foreach my $orig_element ( @{$orig} ) {
+      my $score = $self->_calculate_score( $orig_element, $import_element );
+      if ($score > $best_candidate->{score}) {
+        $best_candidate->{position} = $position;
+        $best_candidate->{score} = $score;
       }
+      $position++;
     }
-    
-    @array = grep { !exists $hash->{$_} } @array;
+    my ($import_key) = ( ref $import_element eq 'HASH' ) ? keys %{$import_element} : ($import_element);
 
-    return keys %{$hash}
-      ? ( scalar(@array)
-            ? [$hash, @array]
-            : $hash
-        )
-      : \@array;
+    if ($best_candidate->{score} == 0 || exists $seen_keys->{$import_key}) {
+      push( @{$orig}, $import_element );
+    } else {
+      my $orig_best = $orig->[$best_candidate->{position}];
+      # merge orig_best and b_element together and replace original with merged
+      if (ref $orig_best ne 'HASH') {
+        $orig->[$best_candidate->{position}] = $import_element;
+      } elsif (ref $import_element eq 'HASH') {
+        my ($key) = keys %{$orig_best};
+        $orig->[$best_candidate->{position}] = { $key => $self->_merge_attr($orig_best->{$key}, $import_element->{$key}) };
+      }
+    }
+    $seen_keys->{$import_key} = 1; # don't merge the same key twice
   }
+
+  return $orig;
 }
 
 sub result_source {
@@ -1914,32 +3065,49 @@ See L<DBIx::Class::Schema/throw_exception> for details.
 
 sub throw_exception {
   my $self=shift;
-  $self->_source_handle->schema->throw_exception(@_);
+  if (ref $self && $self->_source_handle->schema) {
+    $self->_source_handle->schema->throw_exception(@_)
+  } else {
+    croak(@_);
+  }
+
 }
 
 # XXX: FIXME: Attributes docs need clearing up
 
 =head1 ATTRIBUTES
 
-The resultset takes various attributes that modify its behavior. Here's an
-overview of them:
+Attributes are used to refine a ResultSet in various ways when
+searching for data. They can be passed to any method which takes an
+C<\%attrs> argument. See L</search>, L</search_rs>, L</find>,
+L</count>.
+
+These are in no particular order:
 
 =head2 order_by
 
 =over 4
 
-=item Value: ($order_by | \@order_by)
+=item Value: ( $order_by | \@order_by | \%order_by )
 
 =back
 
-Which column(s) to order the results by. This is currently passed
-through directly to SQL, so you can give e.g. C<year DESC> for a
-descending order on the column `year'.
+Which column(s) to order the results by. If a single column name, or
+an arrayref of names is supplied, the argument is passed through
+directly to SQL. The hashref syntax allows for connection-agnostic
+specification of ordering direction:
+
+ For descending order:
+
+  order_by => { -desc => [qw/col1 col2 col3/] }
+
+ For explicit ascending order:
+
+  order_by => { -asc => 'col' }
 
-Please note that if you have C<quote_char> enabled (see
-L<DBIx::Class::Storage::DBI/connect_info>) you will need to do C<\'year DESC' > to
-specify an order. (The scalar ref causes it to be passed as raw sql to the DB,
-so you will need to manually quote things as appropriate.)
+The old scalarref syntax (i.e. order_by => \'year DESC') is still
+supported, although you are strongly encouraged to use the hashref
+syntax as outlined above.
 
 =head2 columns
 
@@ -1949,12 +3117,15 @@ so you will need to manually quote things as appropriate.)
 
 =back
 
-Shortcut to request a particular set of columns to be retrieved.  Adds
-C<me.> 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.)
+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
+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.)
 
-=head2 include_columns
+=head2 +columns
 
 =over 4
 
@@ -1962,10 +3133,13 @@ use the C<cols> attribute, as in earlier versions of DBIC.)
 
 =back
 
-Shortcut to include additional columns in the returned results - for example
+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:-
 
   $schema->resultset('CD')->search(undef, {
-    include_columns => ['artist.name'],
+    '+columns' => ['artist.name'],
     join => ['artist']
   });
 
@@ -1974,6 +3148,16 @@ 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.
 
+=head2 include_columns
+
+=over 4
+
+=item Value: \@columns
+
+=back
+
+Deprecated.  Acts as a synonym for L</+columns> for backward compatibility.
+
 =head2 select
 
 =over 4
@@ -2003,7 +3187,7 @@ return a column named C<count(employeeid)> in the above example.
 =over 4
 
 Indicates additional columns to be selected from storage.  Works the same as
-L<select> but adds columns to the selection.
+L</select> but adds columns to the selection.
 
 =back
 
@@ -2011,7 +3195,7 @@ L<select> but adds columns to the selection.
 
 =over 4
 
-Indicates additional column names for those added via L<+select>.
+Indicates additional column names for those added via L</+select>. See L</as>.
 
 =back
 
@@ -2023,13 +3207,12 @@ Indicates additional column names for those added via L<+select>.
 
 =back
 
-Indicates column names for object inflation. That is, c< as >
+Indicates column names for object inflation. That is, C<as>
 indicates the name that the column can be accessed as via the
 C<get_column> method (or via the object accessor, B<if one already
-exists>).  It has nothing to do with the SQL code C< SELECT foo AS bar
->.
+exists>).  It has nothing to do with the SQL code C<SELECT foo AS bar>.
 
-The C< as > attribute is used in conjunction with C<select>,
+The C<as> attribute is used in conjunction with C<select>,
 usually when C<select> contains one or more function or stored
 procedure names:
 
@@ -2104,19 +3287,19 @@ For example:
     }
   );
 
-You need to use the relationship (not the table) name in  conditions, 
-because they are aliased as such. The current table is aliased as "me", so 
+You need to use the relationship (not the table) name in  conditions,
+because they are aliased as such. The current table is aliased as "me", so
 you need to use me.column_name in order to avoid ambiguity. For example:
 
-  # Get CDs from 1984 with a 'Foo' track 
+  # Get CDs from 1984 with a 'Foo' track
   my $rs = $schema->resultset('CD')->search(
-    { 
+    {
       'me.year' => 1984,
       'tracks.name' => 'Foo'
     },
     { join => 'tracks' }
   );
-  
+
 If the same join is supplied twice, it will be aliased to <rel>_2 (and
 similarly for a third time). For e.g.
 
@@ -2133,6 +3316,8 @@ to Earth' and a cd with title 'Popular'.
 If you want to fetch related objects from other tables as well, see C<prefetch>
 below.
 
+For more help on using joins with search, see L<DBIx::Class::Manual::Joining>.
+
 =head2 prefetch
 
 =over 4
@@ -2141,10 +3326,11 @@ below.
 
 =back
 
-Contains one or more relationships that should be fetched along with the main
-query (when they are accessed afterwards they will have already been
-"prefetched").  This is useful for when you know you will need the related
-objects, because it saves at least one query:
+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,
@@ -2166,13 +3352,27 @@ 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. If you're prefetching to
-depth (e.g. { cd => { artist => 'label' } or similar), you'll need to
-specify the join as well.
+for a C<join> attribute in the above search.
 
 C<prefetch> can be used with the following relationship types: C<belongs_to>,
 C<has_one> (or if you're using C<add_relationship>, any relationship declared
-with an accessor type of 'single' or 'filter').
+with an accessor type of 'single' or 'filter'). A more complex example that
+prefetches an artists cds, the tracks on those cds, and the tags associted
+with that artist is given below (assuming many-to-many from artists to tags):
+
+ my $rs = $schema->resultset('Artist')->search(
+   undef,
+   {
+     prefetch => [
+       { cds => 'tracks' },
+       { artist_tags => 'tags' }
+     ]
+   }
+ );
+
+
+B<NOTE:> If you specify a C<prefetch> attribute, the C<join> and C<select>
+attributes will be ignored.
 
 =head2 page
 
@@ -2186,7 +3386,11 @@ Makes the resultset paged and specifies the page to retrieve. Effectively
 identical to creating a non-pages resultset and then calling ->page($page)
 on it.
 
-If L<rows> attribute is not specified it defualts to 10 rows per page.
+If L<rows> attribute is not specified it defaults to 10 rows per page.
+
+When you have a paged resultset, L</count> will only return the number
+of rows in the page. To get the total, use the L</pager> and call
+C<total_entries> on it.
 
 =head2 rows
 
@@ -2385,6 +3589,81 @@ with a father in the person table, we could explicitly use C<INNER JOIN>:
     # SELECT child.* FROM person child
     # INNER JOIN person father ON child.father_id = father.id
 
+You can select from a subquery by passing a resultset to from as follows.
+
+    $schema->resultset('Artist')->search( 
+        undef, 
+        {   alias => 'artist2',
+            from  => [ { artist2 => $artist_rs->as_query } ],
+        } );
+
+    # and you'll get sql like this..
+    # SELECT artist2.artistid, artist2.name, artist2.rank, artist2.charfield FROM 
+    #   ( SELECT me.artistid, me.name, me.rank, me.charfield FROM artists me ) artist2
+
+If you need to express really complex joins, you
+can supply literal SQL to C<from> via a scalar reference. In this case
+the contents of the scalar will replace the table name associated with the
+resultsource.
+
+WARNING: This technique might very well not work as expected on chained
+searches - you have been warned.
+
+    # Assuming the Event resultsource is defined as:
+
+        MySchema::Event->add_columns (
+            sequence => {
+                data_type => 'INT',
+                is_auto_increment => 1,
+            },
+            location => {
+                data_type => 'INT',
+            },
+            type => {
+                data_type => 'INT',
+            },
+        );
+        MySchema::Event->set_primary_key ('sequence');
+
+    # This will get back the latest event for every location. The column
+    # selector is still provided by DBIC, all we do is add a JOIN/WHERE
+    # combo to limit the resultset
+
+    $rs = $schema->resultset('Event');
+    $table = $rs->result_source->name;
+    $latest = $rs->search (
+        undef,
+        { from => \ "
+            (SELECT e1.* FROM $table e1
+                JOIN $table e2
+                    ON e1.location = e2.location
+                    AND e1.sequence < e2.sequence
+                WHERE e2.sequence is NULL
+            ) me",
+        },
+    );
+
+    # Equivalent SQL (with the DBIC chunks added):
+
+    SELECT me.sequence, me.location, me.type FROM
+       (SELECT e1.* FROM events e1
+           JOIN events e2
+               ON e1.location = e2.location
+               AND e1.sequence < e2.sequence
+           WHERE e2.sequence is NULL
+       ) me;
+
+=head2 for
+
+=over 4
+
+=item Value: ( 'update' | 'shared' )
+
+=back
+
+Set to 'update' for a SELECT ... FOR UPDATE or 'shared' for a SELECT
+... FOR SHARED.
+
 =cut
 
 1;