lib/DBIx/Class/ResultSet.pm

   1 package DBIx::Class::ResultSet;
   2
   3 use strict;
   4 use warnings;
   5 use base qw/DBIx::Class/;
   6 use Carp::Clan qw/^DBIx::Class/;
   7 use DBIx::Class::Exception;
   8 use Data::Page;
   9 use DBIx::Class::ResultSetColumn;
  10 use DBIx::Class::ResultSourceHandle;
  11 use Hash::Merge ();
  12 use Scalar::Util qw/blessed weaken/;
  13 use Try::Tiny;
  14 use Storable qw/nfreeze thaw/;
  15
  16 # not importing first() as it will clash with our own method
  17 use List::Util ();
  18
  19 use namespace::clean;
  20
  21
  22 BEGIN {
  23   # De-duplication in _merge_attr() is disabled, but left in for reference
  24   # (the merger is used for other things that ought not to be de-duped)
  25   *__HM_DEDUP = sub () { 0 };
  26 }
  27
  28 use overload
  29         '0+'     => "count",
  30         'bool'   => "_bool",
  31         fallback => 1;
  32
  33 __PACKAGE__->mk_group_accessors('simple' => qw/_result_class result_source/);
  34
  35 =head1 NAME
  36
  37 DBIx::Class::ResultSet - Represents a query used for fetching a set of results.
  38
  39 =head1 SYNOPSIS
  40
  41   my $users_rs   = $schema->resultset('User');
  42   while( $user = $users_rs->next) {
  43     print $user->username;
  44   }
  45
  46   my $registered_users_rs   = $schema->resultset('User')->search({ registered => 1 });
  47   my @cds_in_2005 = $schema->resultset('CD')->search({ year => 2005 })->all();
  48
  49 =head1 DESCRIPTION
  50
  51 A ResultSet is an object which stores a set of conditions representing
  52 a query. It is the backbone of DBIx::Class (i.e. the really
  53 important/useful bit).
  54
  55 No SQL is executed on the database when a ResultSet is created, it
  56 just stores all the conditions needed to create the query.
  57
  58 A basic ResultSet representing the data of an entire table is returned
  59 by calling C<resultset> on a L<DBIx::Class::Schema> and passing in a
  60 L<Source|DBIx::Class::Manual::Glossary/Source> name.
  61
  62   my $users_rs = $schema->resultset('User');
  63
  64 A new ResultSet is returned from calling L</search> on an existing
  65 ResultSet. The new one will contain all the conditions of the
  66 original, plus any new conditions added in the C<search> call.
  67
  68 A ResultSet also incorporates an implicit iterator. L</next> and L</reset>
  69 can be used to walk through all the L<DBIx::Class::Row>s the ResultSet
  70 represents.
  71
  72 The query that the ResultSet represents is B<only> executed against
  73 the database when these methods are called:
  74 L</find>, L</next>, L</all>, L</first>, L</single>, L</count>.
  75
  76 If a resultset is used in a numeric context it returns the L</count>.
  77 However, if it is used in a boolean context it is B<always> true.  So if
  78 you want to check if a resultset has any results, you must use C<if $rs
  79 != 0>.
  80
  81 =head1 EXAMPLES
  82
  83 =head2 Chaining resultsets
  84
  85 Let's say you've got a query that needs to be run to return some data
  86 to the user. But, you have an authorization system in place that
  87 prevents certain users from seeing certain information. So, you want
  88 to construct the basic query in one method, but add constraints to it in
  89 another.
  90
  91   sub get_data {
  92     my $self = shift;
  93     my $request = $self->get_request; # Get a request object somehow.
  94     my $schema = $self->get_schema;   # Get the DBIC schema object somehow.
  95
  96     my $cd_rs = $schema->resultset('CD')->search({
  97       title => $request->param('title'),
  98       year => $request->param('year'),
  99     });
 100
 101     $self->apply_security_policy( $cd_rs );
 102
 103     return $cd_rs->all();
 104   }
 105
 106   sub apply_security_policy {
 107     my $self = shift;
 108     my ($rs) = @_;
 109
 110     return $rs->search({
 111       subversive => 0,
 112     });
 113   }
 114
 115 =head3 Resolving conditions and attributes
 116
 117 When a resultset is chained from another resultset, conditions and
 118 attributes with the same keys need resolving.
 119
 120 L</join>, L</prefetch>, L</+select>, L</+as> attributes are merged
 121 into the existing ones from the original resultset.
 122
 123 The L</where> and L</having> attributes, and any search conditions, are
 124 merged with an SQL C<AND> to the existing condition from the original
 125 resultset.
 126
 127 All other attributes are overridden by any new ones supplied in the
 128 search attributes.
 129
 130 =head2 Multiple queries
 131
 132 Since a resultset just defines a query, you can do all sorts of
 133 things with it with the same object.
 134
 135   # Don't hit the DB yet.
 136   my $cd_rs = $schema->resultset('CD')->search({
 137     title => 'something',
 138     year => 2009,
 139   });
 140
 141   # Each of these hits the DB individually.
 142   my $count = $cd_rs->count;
 143   my $most_recent = $cd_rs->get_column('date_released')->max();
 144   my @records = $cd_rs->all;
 145
 146 And it's not just limited to SELECT statements.
 147
 148   $cd_rs->delete();
 149
 150 This is even cooler:
 151
 152   $cd_rs->create({ artist => 'Fred' });
 153
 154 Which is the same as:
 155
 156   $schema->resultset('CD')->create({
 157     title => 'something',
 158     year => 2009,
 159     artist => 'Fred'
 160   });
 161
 162 See: L</search>, L</count>, L</get_column>, L</all>, L</create>.
 163
 164 =head1 METHODS
 165
 166 =head2 new
 167
 168 =over 4
 169
 170 =item Arguments: $source, \%$attrs
 171
 172 =item Return Value: $rs
 173
 174 =back
 175
 176 The resultset constructor. Takes a source object (usually a
 177 L<DBIx::Class::ResultSourceProxy::Table>) and an attribute hash (see
 178 L</ATTRIBUTES> below).  Does not perform any queries -- these are
 179 executed as needed by the other methods.
 180
 181 Generally you won't need to construct a resultset manually.  You'll
 182 automatically get one from e.g. a L</search> called in scalar context:
 183
 184   my $rs = $schema->resultset('CD')->search({ title => '100th Window' });
 185
 186 IMPORTANT: If called on an object, proxies to new_result instead so
 187
 188   my $cd = $schema->resultset('CD')->new({ title => 'Spoon' });
 189
 190 will return a CD object, not a ResultSet.
 191
 192 =cut
 193
 194 sub new {
 195   my $class = shift;
 196   return $class->new_result(@_) if ref $class;
 197
 198   my ($source, $attrs) = @_;
 199   $source = $source->resolve
 200     if $source->isa('DBIx::Class::ResultSourceHandle');
 201   $attrs = { %{$attrs||{}} };
 202
 203   if ($attrs->{page}) {
 204     $attrs->{rows} ||= 10;
 205   }
 206
 207   $attrs->{alias} ||= 'me';
 208
 209   # Creation of {} and bless separated to mitigate RH perl bug
 210   # see https://bugzilla.redhat.com/show_bug.cgi?id=196836
 211   my $self = {
 212     result_source => $source,
 213     cond => $attrs->{where},
 214     pager => undef,
 215     attrs => $attrs,
 216   };
 217
 218   bless $self, $class;
 219
 220   $self->result_class(
 221     $attrs->{result_class} || $source->result_class
 222   );
 223
 224   return $self;
 225 }
 226
 227 =head2 search
 228
 229 =over 4
 230
 231 =item Arguments: $cond, \%attrs?
 232
 233 =item Return Value: $resultset (scalar context), @row_objs (list context)
 234
 235 =back
 236
 237   my @cds    = $cd_rs->search({ year => 2001 }); # "... WHERE year = 2001"
 238   my $new_rs = $cd_rs->search({ year => 2005 });
 239
 240   my $new_rs = $cd_rs->search([ { year => 2005 }, { year => 2004 } ]);
 241                  # year = 2005 OR year = 2004
 242
 243 If you need to pass in additional attributes but no additional condition,
 244 call it as C<search(undef, \%attrs)>.
 245
 246   # "SELECT name, artistid FROM $artist_table"
 247   my @all_artists = $schema->resultset('Artist')->search(undef, {
 248     columns => [qw/name artistid/],
 249   });
 250
 251 For a list of attributes that can be passed to C<search>, see
 252 L</ATTRIBUTES>. For more examples of using this function, see
 253 L<Searching|DBIx::Class::Manual::Cookbook/Searching>. For a complete
 254 documentation for the first argument, see L<SQL::Abstract>.
 255
 256 For more help on using joins with search, see L<DBIx::Class::Manual::Joining>.
 257
 258 =head3 CAVEAT
 259
 260 Note that L</search> does not process/deflate any of the values passed in the
 261 L<SQL::Abstract>-compatible search condition structure. This is unlike other
 262 condition-bound methods L</new>, L</create> and L</find>. The user must ensure
 263 manually that any value passed to this method will stringify to something the
 264 RDBMS knows how to deal with. A notable example is the handling of L<DateTime>
 265 objects, for more info see:
 266 L<DBIx::Class::Manual::Cookbook/Formatting_DateTime_objects_in_queries>.
 267
 268 =cut
 269
 270 sub search {
 271   my $self = shift;
 272   my $rs = $self->search_rs( @_ );
 273
 274   if (wantarray) {
 275     return $rs->all;
 276   }
 277   elsif (defined wantarray) {
 278     return $rs;
 279   }
 280   else {
 281     # we can be called by a relationship helper, which in
 282     # turn may be called in void context due to some braindead
 283     # overload or whatever else the user decided to be clever
 284     # at this particular day. Thus limit the exception to
 285     # external code calls only
 286     $self->throw_exception ('->search is *not* a mutator, calling it in void context makes no sense')
 287       if (caller)[0] !~ /^\QDBIx::Class::/;
 288
 289     return ();
 290   }
 291 }
 292
 293 =head2 search_rs
 294
 295 =over 4
 296
 297 =item Arguments: $cond, \%attrs?
 298
 299 =item Return Value: $resultset
 300
 301 =back
 302
 303 This method does the same exact thing as search() except it will
 304 always return a resultset, even in list context.
 305
 306 =cut
 307
 308 my $callsites_warned;
 309 sub search_rs {
 310   my $self = shift;
 311
 312   # Special-case handling for (undef, undef).
 313   if ( @_ == 2 && !defined $_[1] && !defined $_[0] ) {
 314     @_ = ();
 315   }
 316
 317   my $call_attrs = {};
 318   if (@_ > 1) {
 319     if (ref $_[-1] eq 'HASH') {
 320       # copy for _normalize_selection
 321       $call_attrs = { %{ pop @_ } };
 322     }
 323     elsif (! defined $_[-1] ) {
 324       pop @_;   # search({}, undef)
 325     }
 326   }
 327
 328   # see if we can keep the cache (no $rs changes)
 329   my $cache;
 330   my %safe = (alias => 1, cache => 1);
 331   if ( ! List::Util::first { !$safe{$_} } keys %$call_attrs and (
 332     ! defined $_[0]
 333       or
 334     ref $_[0] eq 'HASH' && ! keys %{$_[0]}
 335       or
 336     ref $_[0] eq 'ARRAY' && ! @{$_[0]}
 337   )) {
 338     $cache = $self->get_cache;
 339   }
 340
 341   my $rsrc = $self->result_source;
 342
 343   my $old_attrs = { %{$self->{attrs}} };
 344   my $old_having = delete $old_attrs->{having};
 345   my $old_where = delete $old_attrs->{where};
 346
 347   my $new_attrs = { %$old_attrs };
 348
 349   # take care of call attrs (only if anything is changing)
 350   if (keys %$call_attrs) {
 351
 352     $self->throw_exception ('_trailing_select is not a public attribute - do not use it in search()')
 353       if ( exists $call_attrs->{_trailing_select} or exists $call_attrs->{'+_trailing_select'} );
 354
 355     my @selector_attrs = qw/select as columns cols +select +as +columns include_columns _trailing_select +_trailing_select/;
 356
 357     # Normalize the selector list (operates on the passed-in attr structure)
 358     # Need to do it on every chain instead of only once on _resolved_attrs, in
 359     # order to separate 'as'-ed from blind 'select's
 360     $self->_normalize_selection ($call_attrs);
 361
 362     # start with blind overwriting merge, exclude selector attrs
 363     $new_attrs = { %{$old_attrs}, %{$call_attrs} };
 364     delete @{$new_attrs}{@selector_attrs};
 365
 366     # reset the current selector list if new selectors are supplied
 367     if (List::Util::first { exists $call_attrs->{$_} } qw/columns cols select as/) {
 368       delete @{$old_attrs}{@selector_attrs};
 369     }
 370
 371     for (@selector_attrs) {
 372       $new_attrs->{$_} = $self->_merge_attr($old_attrs->{$_}, $call_attrs->{$_})
 373         if ( exists $old_attrs->{$_} or exists $call_attrs->{$_} );
 374     }
 375
 376     # older deprecated name, use only if {columns} is not there
 377     if (my $c = delete $new_attrs->{cols}) {
 378       if ($new_attrs->{columns}) {
 379         carp "Resultset specifies both the 'columns' and the legacy 'cols' attributes - ignoring 'cols'";
 380       }
 381       else {
 382         $new_attrs->{columns} = $c;
 383       }
 384     }
 385
 386
 387     # join/prefetch use their own crazy merging heuristics
 388     foreach my $key (qw/join prefetch/) {
 389       $new_attrs->{$key} = $self->_merge_joinpref_attr($old_attrs->{$key}, $call_attrs->{$key})
 390         if exists $call_attrs->{$key};
 391     }
 392
 393     # stack binds together
 394     $new_attrs->{bind} = [ @{ $old_attrs->{bind} || [] }, @{ $call_attrs->{bind} || [] } ];
 395   }
 396
 397
 398   # rip apart the rest of @_, parse a condition
 399   my $call_cond = do {
 400
 401     if (ref $_[0] eq 'HASH') {
 402       (keys %{$_[0]}) ? $_[0] : undef
 403     }
 404     elsif (@_ == 1) {
 405       $_[0]
 406     }
 407     elsif (@_ % 2) {
 408       $self->throw_exception('Odd number of arguments to search')
 409     }
 410     else {
 411       +{ @_ }
 412     }
 413
 414   } if @_;
 415
 416   if( @_ > 1 and ! $rsrc->result_class->isa('DBIx::Class::CDBICompat') ) {
 417     # determine callsite obeying Carp::Clan rules (fucking ugly but don't have better ideas)
 418     my $callsite = do {
 419       my $w;
 420       local $SIG{__WARN__} = sub { $w = shift };
 421       carp;
 422       $w
 423     };
 424     carp 'search( %condition ) is deprecated, use search( \%condition ) instead'
 425       unless $callsites_warned->{$callsite}++;
 426   }
 427
 428   for ($old_where, $call_cond) {
 429     if (defined $_) {
 430       $new_attrs->{where} = $self->_stack_cond (
 431         $_, $new_attrs->{where}
 432       );
 433     }
 434   }
 435
 436   if (defined $old_having) {
 437     $new_attrs->{having} = $self->_stack_cond (
 438       $old_having, $new_attrs->{having}
 439     )
 440   }
 441
 442   my $rs = (ref $self)->new($rsrc, $new_attrs);
 443
 444   $rs->set_cache($cache) if ($cache);
 445
 446   return $rs;
 447 }
 448
 449 sub _normalize_selection {
 450   my ($self, $attrs) = @_;
 451
 452   # legacy syntax
 453   $attrs->{'+columns'} = $self->_merge_attr($attrs->{'+columns'}, delete $attrs->{include_columns})
 454     if exists $attrs->{include_columns};
 455
 456   # Keep the X vs +X separation until _resolved_attrs time - this allows to
 457   # delay the decision on whether to use a default select list ($rsrc->columns)
 458   # allowing stuff like the remove_columns helper to work
 459   #
 460   # select/as +select/+as pairs need special handling - the amount of select/as
 461   # elements in each pair does *not* have to be equal (think multicolumn
 462   # selectors like distinct(foo, bar) ). If the selector is bare (no 'as'
 463   # supplied at all) - try to infer the alias, either from the -as parameter
 464   # of the selector spec, or use the parameter whole if it looks like a column
 465   # name (ugly legacy heuristic). If all fails - leave the selector bare (which
 466   # is ok as well), but transport it over a separate attribute to make sure it is
 467   # the last thing in the select list, thus unable to throw off the corresponding
 468   # 'as' chain
 469   for my $pref ('', '+') {
 470
 471     my ($sel, $as) = map {
 472       my $key = "${pref}${_}";
 473
 474       my $val = [ ref $attrs->{$key} eq 'ARRAY'
 475         ? @{$attrs->{$key}}
 476         : $attrs->{$key} || ()
 477       ];
 478       delete $attrs->{$key};
 479       $val;
 480     } qw/select as/;
 481
 482     if (! @$as and ! @$sel ) {
 483       next;
 484     }
 485     elsif (@$as and ! @$sel) {
 486       $self->throw_exception(
 487         "Unable to handle ${pref}as specification (@$as) without a corresponding ${pref}select"
 488       );
 489     }
 490     elsif( ! @$as ) {
 491       # no as part supplied at all - try to deduce
 492       # if any @$as has been supplied we assume the user knows what (s)he is doing
 493       # and blindly keep stacking up pieces
 494       my (@new_sel, @new_trailing);
 495       for (@$sel) {
 496         if ( ref $_ eq 'HASH' and exists $_->{-as} ) {
 497           push @$as, $_->{-as};
 498           push @new_sel, $_;
 499         }
 500         # assume any plain no-space, no-parenthesis string to be a column spec
 501         # FIXME - this is retarded but is necessary to support shit like 'count(foo)'
 502         elsif ( ! ref $_ and $_ =~ /^ [^\s\(\)]+ $/x) {
 503           push @$as, $_;
 504           push @new_sel, $_;
 505         }
 506         # if all else fails - shove the selection to the trailing stack and move on
 507         else {
 508           push @new_trailing, $_;
 509         }
 510       }
 511
 512       @$sel = @new_sel;
 513       $attrs->{"${pref}_trailing_select"} = $self->_merge_attr($attrs->{"${pref}_trailing_select"}, \@new_trailing)
 514         if @new_trailing;
 515     }
 516     elsif (@$as < @$sel) {
 517       $self->throw_exception(
 518         "Unable to handle an ${pref}as specification (@$as) with less elements than the corresponding ${pref}select"
 519       );
 520     }
 521
 522     # now see what the result for this pair looks like:
 523     if (@$as == @$sel) {
 524
 525       # if balanced - treat as a columns entry
 526       $attrs->{"${pref}columns"} = $self->_merge_attr(
 527         $attrs->{"${pref}columns"},
 528         [ map { +{ $as->[$_] => $sel->[$_] } } ( 0 .. $#$as ) ]
 529       );
 530     }
 531     else {
 532       # unbalanced - shove in select/as, not subject to deduplication in _resolved_attrs
 533       $attrs->{"${pref}select"} = $self->_merge_attr($attrs->{"${pref}select"}, $sel);
 534       $attrs->{"${pref}as"} = $self->_merge_attr($attrs->{"${pref}as"}, $as);
 535     }
 536   }
 537
 538 }
 539
 540 sub _stack_cond {
 541   my ($self, $left, $right) = @_;
 542   if (defined $left xor defined $right) {
 543     return defined $left ? $left : $right;
 544   }
 545   elsif (defined $left) {
 546     return { -and => [ map
 547       { ref $_ eq 'ARRAY' ? [ -or => $_ ] : $_ }
 548       ($left, $right)
 549     ]};
 550   }
 551
 552   return undef;
 553 }
 554
 555 =head2 search_literal
 556
 557 =over 4
 558
 559 =item Arguments: $sql_fragment, @bind_values
 560
 561 =item Return Value: $resultset (scalar context), @row_objs (list context)
 562
 563 =back
 564
 565   my @cds   = $cd_rs->search_literal('year = ? AND title = ?', qw/2001 Reload/);
 566   my $newrs = $artist_rs->search_literal('name = ?', 'Metallica');
 567
 568 Pass a literal chunk of SQL to be added to the conditional part of the
 569 resultset query.
 570
 571 CAVEAT: C<search_literal> is provided for Class::DBI compatibility and should
 572 only be used in that context. C<search_literal> is a convenience method.
 573 It is equivalent to calling $schema->search(\[]), but if you want to ensure
 574 columns are bound correctly, use C<search>.
 575
 576 Example of how to use C<search> instead of C<search_literal>
 577
 578   my @cds = $cd_rs->search_literal('cdid = ? AND (artist = ? OR artist = ?)', (2, 1, 2));
 579   my @cds = $cd_rs->search(\[ 'cdid = ? AND (artist = ? OR artist = ?)', [ 'cdid', 2 ], [ 'artist', 1 ], [ 'artist', 2 ] ]);
 580
 581
 582 See L<DBIx::Class::Manual::Cookbook/Searching> and
 583 L<DBIx::Class::Manual::FAQ/Searching> for searching techniques that do not
 584 require C<search_literal>.
 585
 586 =cut
 587
 588 sub search_literal {
 589   my ($self, $sql, @bind) = @_;
 590   my $attr;
 591   if ( @bind && ref($bind[-1]) eq 'HASH' ) {
 592     $attr = pop @bind;
 593   }
 594   return $self->search(\[ $sql, map [ __DUMMY__ => $_ ], @bind ], ($attr || () ));
 595 }
 596
 597 =head2 find
 598
 599 =over 4
 600
 601 =item Arguments: \%columns_values | @pk_values, \%attrs?
 602
 603 =item Return Value: $row_object | undef
 604
 605 =back
 606
 607 Finds and returns a single row based on supplied criteria. Takes either a
 608 hashref with the same format as L</create> (including inference of foreign
 609 keys from related objects), or a list of primary key values in the same
 610 order as the L<primary columns|DBIx::Class::ResultSource/primary_columns>
 611 declaration on the L</result_source>.
 612
 613 In either case an attempt is made to combine conditions already existing on
 614 the resultset with the condition passed to this method.
 615
 616 To aid with preparing the correct query for the storage you may supply the
 617 C<key> attribute, which is the name of a
 618 L<unique constraint|DBIx::Class::ResultSource/add_unique_constraint> (the
 619 unique constraint corresponding to the
 620 L<primary columns|DBIx::Class::ResultSource/primary_columns> is always named
 621 C<primary>). If the C<key> attribute has been supplied, and DBIC is unable
 622 to construct a query that satisfies the named unique constraint fully (
 623 non-NULL values for each column member of the constraint) an exception is
 624 thrown.
 625
 626 If no C<key> is specified, the search is carried over all unique constraints
 627 which are fully defined by the available condition.
 628
 629 If no such constraint is found, C<find> currently defaults to a simple
 630 C<< search->(\%column_values) >> which may or may not do what you expect.
 631 Note that this fallback behavior may be deprecated in further versions. If
 632 you need to search with arbitrary conditions - use L</search>. If the query
 633 resulting from this fallback produces more than one row, a warning to the
 634 effect is issued, though only the first row is constructed and returned as
 635 C<$row_object>.
 636
 637 In addition to C<key>, L</find> recognizes and applies standard
 638 L<resultset attributes|/ATTRIBUTES> in the same way as L</search> does.
 639
 640 Note that if you have extra concerns about the correctness of the resulting
 641 query you need to specify the C<key> attribute and supply the entire condition
 642 as an argument to find (since it is not always possible to perform the
 643 combination of the resultset condition with the supplied one, especially if
 644 the resultset condition contains literal sql).
 645
 646 For example, to find a row by its primary key:
 647
 648   my $cd = $schema->resultset('CD')->find(5);
 649
 650 You can also find a row by a specific unique constraint:
 651
 652   my $cd = $schema->resultset('CD')->find(
 653     {
 654       artist => 'Massive Attack',
 655       title  => 'Mezzanine',
 656     },
 657     { key => 'cd_artist_title' }
 658   );
 659
 660 See also L</find_or_create> and L</update_or_create>.
 661
 662 =cut
 663
 664 sub find {
 665   my $self = shift;
 666   my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
 667
 668   my $rsrc = $self->result_source;
 669
 670   # Parse out the condition from input
 671   my $call_cond;
 672   if (ref $_[0] eq 'HASH') {
 673     $call_cond = { %{$_[0]} };
 674   }
 675   else {
 676     my $constraint = exists $attrs->{key} ? $attrs->{key} : 'primary';
 677     my @c_cols = $rsrc->unique_constraint_columns($constraint);
 678
 679     $self->throw_exception(
 680       "No constraint columns, maybe a malformed '$constraint' constraint?"
 681     ) unless @c_cols;
 682
 683     $self->throw_exception (
 684       'find() expects either a column/value hashref, or a list of values '
 685     . "corresponding to the columns of the specified unique constraint '$constraint'"
 686     ) unless @c_cols == @_;
 687
 688     $call_cond = {};
 689     @{$call_cond}{@c_cols} = @_;
 690   }
 691
 692   my %related;
 693   for my $key (keys %$call_cond) {
 694     if (
 695       my $keyref = ref($call_cond->{$key})
 696         and
 697       my $relinfo = $rsrc->relationship_info($key)
 698     ) {
 699       my $val = delete $call_cond->{$key};
 700
 701       next if $keyref eq 'ARRAY'; # has_many for multi_create
 702
 703       my $rel_q = $rsrc->_resolve_condition(
 704         $relinfo->{cond}, $val, $key
 705       );
 706       die "Can't handle complex relationship conditions in find" if ref($rel_q) ne 'HASH';
 707       @related{keys %$rel_q} = values %$rel_q;
 708     }
 709   }
 710
 711   # relationship conditions take precedence (?)
 712   @{$call_cond}{keys %related} = values %related;
 713
 714   my $alias = exists $attrs->{alias} ? $attrs->{alias} : $self->{attrs}{alias};
 715   my $final_cond;
 716   if (exists $attrs->{key}) {
 717     $final_cond = $self->_qualify_cond_columns (
 718
 719       $self->_build_unique_cond (
 720         $attrs->{key},
 721         $call_cond,
 722       ),
 723
 724       $alias,
 725     );
 726   }
 727   elsif ($self->{attrs}{accessor} and $self->{attrs}{accessor} eq 'single') {
 728     # This means that we got here after a merger of relationship conditions
 729     # in ::Relationship::Base::search_related (the row method), and furthermore
 730     # the relationship is of the 'single' type. This means that the condition
 731     # provided by the relationship (already attached to $self) is sufficient,
 732     # as there can be only one row in the database that would satisfy the
 733     # relationship
 734   }
 735   else {
 736     # no key was specified - fall down to heuristics mode:
 737     # run through all unique queries registered on the resultset, and
 738     # 'OR' all qualifying queries together
 739     my (@unique_queries, %seen_column_combinations);
 740     for my $c_name ($rsrc->unique_constraint_names) {
 741       next if $seen_column_combinations{
 742         join "\x00", sort $rsrc->unique_constraint_columns($c_name)
 743       }++;
 744
 745       push @unique_queries, try {
 746         $self->_build_unique_cond ($c_name, $call_cond)
 747       } || ();
 748     }
 749
 750     $final_cond = @unique_queries
 751       ? [ map { $self->_qualify_cond_columns($_, $alias) } @unique_queries ]
 752       : $self->_non_unique_find_fallback ($call_cond, $attrs)
 753     ;
 754   }
 755
 756   # Run the query, passing the result_class since it should propagate for find
 757   my $rs = $self->search ($final_cond, {result_class => $self->result_class, %$attrs});
 758   if (keys %{$rs->_resolved_attrs->{collapse}}) {
 759     my $row = $rs->next;
 760     carp "Query returned more than one row" if $rs->next;
 761     return $row;
 762   }
 763   else {
 764     return $rs->single;
 765   }
 766 }
 767
 768 # This is a stop-gap method as agreed during the discussion on find() cleanup:
 769 # http://lists.scsys.co.uk/pipermail/dbix-class/2010-October/009535.html
 770 #
 771 # It is invoked when find() is called in legacy-mode with insufficiently-unique
 772 # condition. It is provided for overrides until a saner way forward is devised
 773 #
 774 # *NOTE* This is not a public method, and it's *GUARANTEED* to disappear down
 775 # the road. Please adjust your tests accordingly to catch this situation early
 776 # DBIx::Class::ResultSet->can('_non_unique_find_fallback') is reasonable
 777 #
 778 # The method will not be removed without an adequately complete replacement
 779 # for strict-mode enforcement
 780 sub _non_unique_find_fallback {
 781   my ($self, $cond, $attrs) = @_;
 782
 783   return $self->_qualify_cond_columns(
 784     $cond,
 785     exists $attrs->{alias}
 786       ? $attrs->{alias}
 787       : $self->{attrs}{alias}
 788   );
 789 }
 790
 791
 792 sub _qualify_cond_columns {
 793   my ($self, $cond, $alias) = @_;
 794
 795   my %aliased = %$cond;
 796   for (keys %aliased) {
 797     $aliased{"$alias.$_"} = delete $aliased{$_}
 798       if $_ !~ /\./;
 799   }
 800
 801   return \%aliased;
 802 }
 803
 804 sub _build_unique_cond {
 805   my ($self, $constraint_name, $extra_cond) = @_;
 806
 807   my @c_cols = $self->result_source->unique_constraint_columns($constraint_name);
 808
 809   # combination may fail if $self->{cond} is non-trivial
 810   my ($final_cond) = try {
 811     $self->_merge_with_rscond ($extra_cond)
 812   } catch {
 813     +{ %$extra_cond }
 814   };
 815
 816   # trim out everything not in $columns
 817   $final_cond = { map { $_ => $final_cond->{$_} } @c_cols };
 818
 819   if (my @missing = grep { ! defined $final_cond->{$_} } (@c_cols) ) {
 820     $self->throw_exception( sprintf ( "Unable to satisfy requested constraint '%s', no values for column(s): %s",
 821       $constraint_name,
 822       join (', ', map { "'$_'" } @missing),
 823     ) );
 824   }
 825
 826   return $final_cond;
 827 }
 828
 829 =head2 search_related
 830
 831 =over 4
 832
 833 =item Arguments: $rel, $cond, \%attrs?
 834
 835 =item Return Value: $new_resultset
 836
 837 =back
 838
 839   $new_rs = $cd_rs->search_related('artist', {
 840     name => 'Emo-R-Us',
 841   });
 842
 843 Searches the specified relationship, optionally specifying a condition and
 844 attributes for matching records. See L</ATTRIBUTES> for more information.
 845
 846 =cut
 847
 848 sub search_related {
 849   return shift->related_resultset(shift)->search(@_);
 850 }
 851
 852 =head2 search_related_rs
 853
 854 This method works exactly the same as search_related, except that
 855 it guarantees a resultset, even in list context.
 856
 857 =cut
 858
 859 sub search_related_rs {
 860   return shift->related_resultset(shift)->search_rs(@_);
 861 }
 862
 863 =head2 cursor
 864
 865 =over 4
 866
 867 =item Arguments: none
 868
 869 =item Return Value: $cursor
 870
 871 =back
 872
 873 Returns a storage-driven cursor to the given resultset. See
 874 L<DBIx::Class::Cursor> for more information.
 875
 876 =cut
 877
 878 sub cursor {
 879   my ($self) = @_;
 880
 881   my $attrs = $self->_resolved_attrs_copy;
 882
 883   return $self->{cursor}
 884     ||= $self->result_source->storage->select($attrs->{from}, $attrs->{select},
 885           $attrs->{where},$attrs);
 886 }
 887
 888 =head2 single
 889
 890 =over 4
 891
 892 =item Arguments: $cond?
 893
 894 =item Return Value: $row_object | undef
 895
 896 =back
 897
 898   my $cd = $schema->resultset('CD')->single({ year => 2001 });
 899
 900 Inflates the first result without creating a cursor if the resultset has
 901 any records in it; if not returns C<undef>. Used by L</find> as a lean version
 902 of L</search>.
 903
 904 While this method can take an optional search condition (just like L</search>)
 905 being a fast-code-path it does not recognize search attributes. If you need to
 906 add extra joins or similar, call L</search> and then chain-call L</single> on the
 907 L<DBIx::Class::ResultSet> returned.
 908
 909 =over
 910
 911 =item B<Note>
 912
 913 As of 0.08100, this method enforces the assumption that the preceding
 914 query returns only one row. If more than one row is returned, you will receive
 915 a warning:
 916
 917   Query returned more than one row
 918
 919 In this case, you should be using L</next> or L</find> instead, or if you really
 920 know what you are doing, use the L</rows> attribute to explicitly limit the size
 921 of the resultset.
 922
 923 This method will also throw an exception if it is called on a resultset prefetching
 924 has_many, as such a prefetch implies fetching multiple rows from the database in
 925 order to assemble the resulting object.
 926
 927 =back
 928
 929 =cut
 930
 931 sub single {
 932   my ($self, $where) = @_;
 933   if(@_ > 2) {
 934       $self->throw_exception('single() only takes search conditions, no attributes. You want ->search( $cond, $attrs )->single()');
 935   }
 936
 937   my $attrs = $self->_resolved_attrs_copy;
 938
 939   if (keys %{$attrs->{collapse}}) {
 940     $self->throw_exception(
 941       'single() can not be used on resultsets prefetching has_many. Use find( \%cond ) or next() instead'
 942     );
 943   }
 944
 945   if ($where) {
 946     if (defined $attrs->{where}) {
 947       $attrs->{where} = {
 948         '-and' =>
 949             [ map { ref $_ eq 'ARRAY' ? [ -or => $_ ] : $_ }
 950                $where, delete $attrs->{where} ]
 951       };
 952     } else {
 953       $attrs->{where} = $where;
 954     }
 955   }
 956
 957   my @data = $self->result_source->storage->select_single(
 958     $attrs->{from}, $attrs->{select},
 959     $attrs->{where}, $attrs
 960   );
 961
 962   return (@data ? ($self->_construct_object(@data))[0] : undef);
 963 }
 964
 965
 966 # _collapse_query
 967 #
 968 # Recursively collapse the query, accumulating values for each column.
 969
 970 sub _collapse_query {
 971   my ($self, $query, $collapsed) = @_;
 972
 973   $collapsed ||= {};
 974
 975   if (ref $query eq 'ARRAY') {
 976     foreach my $subquery (@$query) {
 977       next unless ref $subquery;  # -or
 978       $collapsed = $self->_collapse_query($subquery, $collapsed);
 979     }
 980   }
 981   elsif (ref $query eq 'HASH') {
 982     if (keys %$query and (keys %$query)[0] eq '-and') {
 983       foreach my $subquery (@{$query->{-and}}) {
 984         $collapsed = $self->_collapse_query($subquery, $collapsed);
 985       }
 986     }
 987     else {
 988       foreach my $col (keys %$query) {
 989         my $value = $query->{$col};
 990         $collapsed->{$col}{$value}++;
 991       }
 992     }
 993   }
 994
 995   return $collapsed;
 996 }
 997
 998 =head2 get_column
 999
1000 =over 4
1001
1002 =item Arguments: $cond?
1003
1004 =item Return Value: $resultsetcolumn
1005
1006 =back
1007
1008   my $max_length = $rs->get_column('length')->max;
1009
1010 Returns a L<DBIx::Class::ResultSetColumn> instance for a column of the ResultSet.
1011
1012 =cut
1013
1014 sub get_column {
1015   my ($self, $column) = @_;
1016   my $new = DBIx::Class::ResultSetColumn->new($self, $column);
1017   return $new;
1018 }
1019
1020 =head2 search_like
1021
1022 =over 4
1023
1024 =item Arguments: $cond, \%attrs?
1025
1026 =item Return Value: $resultset (scalar context), @row_objs (list context)
1027
1028 =back
1029
1030   # WHERE title LIKE '%blue%'
1031   $cd_rs = $rs->search_like({ title => '%blue%'});
1032
1033 Performs a search, but uses C<LIKE> instead of C<=> as the condition. Note
1034 that this is simply a convenience method retained for ex Class::DBI users.
1035 You most likely want to use L</search> with specific operators.
1036
1037 For more information, see L<DBIx::Class::Manual::Cookbook>.
1038
1039 This method is deprecated and will be removed in 0.09. Use L</search()>
1040 instead. An example conversion is:
1041
1042   ->search_like({ foo => 'bar' });
1043
1044   # Becomes
1045
1046   ->search({ foo => { like => 'bar' } });
1047
1048 =cut
1049
1050 sub search_like {
1051   my $class = shift;
1052   carp (
1053     'search_like() is deprecated and will be removed in DBIC version 0.09.'
1054    .' Instead use ->search({ x => { -like => "y%" } })'
1055    .' (note the outer pair of {}s - they are important!)'
1056   );
1057   my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
1058   my $query = ref $_[0] eq 'HASH' ? { %{shift()} }: {@_};
1059   $query->{$_} = { 'like' => $query->{$_} } for keys %$query;
1060   return $class->search($query, { %$attrs });
1061 }
1062
1063 =head2 slice
1064
1065 =over 4
1066
1067 =item Arguments: $first, $last
1068
1069 =item Return Value: $resultset (scalar context), @row_objs (list context)
1070
1071 =back
1072
1073 Returns a resultset or object list representing a subset of elements from the
1074 resultset slice is called on. Indexes are from 0, i.e., to get the first
1075 three records, call:
1076
1077   my ($one, $two, $three) = $rs->slice(0, 2);
1078
1079 =cut
1080
1081 sub slice {
1082   my ($self, $min, $max) = @_;
1083   my $attrs = {}; # = { %{ $self->{attrs} || {} } };
1084   $attrs->{offset} = $self->{attrs}{offset} || 0;
1085   $attrs->{offset} += $min;
1086   $attrs->{rows} = ($max ? ($max - $min + 1) : 1);
1087   return $self->search(undef, $attrs);
1088   #my $slice = (ref $self)->new($self->result_source, $attrs);
1089   #return (wantarray ? $slice->all : $slice);
1090 }
1091
1092 =head2 next
1093
1094 =over 4
1095
1096 =item Arguments: none
1097
1098 =item Return Value: $result | undef
1099
1100 =back
1101
1102 Returns the next element in the resultset (C<undef> is there is none).
1103
1104 Can be used to efficiently iterate over records in the resultset:
1105
1106   my $rs = $schema->resultset('CD')->search;
1107   while (my $cd = $rs->next) {
1108     print $cd->title;
1109   }
1110
1111 Note that you need to store the resultset object, and call C<next> on it.
1112 Calling C<< resultset('Table')->next >> repeatedly will always return the
1113 first record from the resultset.
1114
1115 =cut
1116
1117 sub next {
1118   my ($self) = @_;
1119   if (my $cache = $self->get_cache) {
1120     $self->{all_cache_position} ||= 0;
1121     return $cache->[$self->{all_cache_position}++];
1122   }
1123   if ($self->{attrs}{cache}) {
1124     delete $self->{pager};
1125     $self->{all_cache_position} = 1;
1126     return ($self->all)[0];
1127   }
1128   if ($self->{stashed_objects}) {
1129     my $obj = shift(@{$self->{stashed_objects}});
1130     delete $self->{stashed_objects} unless @{$self->{stashed_objects}};
1131     return $obj;
1132   }
1133   my @row = (
1134     exists $self->{stashed_row}
1135       ? @{delete $self->{stashed_row}}
1136       : $self->cursor->next
1137   );
1138   return undef unless (@row);
1139   my ($row, @more) = $self->_construct_object(@row);
1140   $self->{stashed_objects} = \@more if @more;
1141   return $row;
1142 }
1143
1144 sub _construct_object {
1145   my ($self, @row) = @_;
1146
1147   my $info = $self->_collapse_result($self->{_attrs}{as}, \@row)
1148     or return ();
1149   my @new = $self->result_class->inflate_result($self->result_source, @$info);
1150   @new = $self->{_attrs}{record_filter}->(@new)
1151     if exists $self->{_attrs}{record_filter};
1152   return @new;
1153 }
1154
1155 sub _collapse_result {
1156   my ($self, $as_proto, $row) = @_;
1157
1158   my @copy = @$row;
1159
1160   # 'foo'         => [ undef, 'foo' ]
1161   # 'foo.bar'     => [ 'foo', 'bar' ]
1162   # 'foo.bar.baz' => [ 'foo.bar', 'baz' ]
1163
1164   my @construct_as = map { [ (/^(?:(.*)\.)?([^.]+)$/) ] } @$as_proto;
1165
1166   my %collapse = %{$self->{_attrs}{collapse}||{}};
1167
1168   my @pri_index;
1169
1170   # if we're doing collapsing (has_many prefetch) we need to grab records
1171   # until the PK changes, so fill @pri_index. if not, we leave it empty so
1172   # we know we don't have to bother.
1173
1174   # the reason for not using the collapse stuff directly is because if you
1175   # had for e.g. two artists in a row with no cds, the collapse info for
1176   # both would be NULL (undef) so you'd lose the second artist
1177
1178   # store just the index so we can check the array positions from the row
1179   # without having to contruct the full hash
1180
1181   if (keys %collapse) {
1182     my %pri = map { ($_ => 1) } $self->result_source->_pri_cols;
1183     foreach my $i (0 .. $#construct_as) {
1184       next if defined($construct_as[$i][0]); # only self table
1185       if (delete $pri{$construct_as[$i][1]}) {
1186         push(@pri_index, $i);
1187       }
1188       last unless keys %pri; # short circuit (Johnny Five Is Alive!)
1189     }
1190   }
1191
1192   # no need to do an if, it'll be empty if @pri_index is empty anyway
1193
1194   my %pri_vals = map { ($_ => $copy[$_]) } @pri_index;
1195
1196   my @const_rows;
1197
1198   do { # no need to check anything at the front, we always want the first row
1199
1200     my %const;
1201
1202     foreach my $this_as (@construct_as) {
1203       $const{$this_as->[0]||''}{$this_as->[1]} = shift(@copy);
1204     }
1205
1206     push(@const_rows, \%const);
1207
1208   } until ( # no pri_index => no collapse => drop straight out
1209       !@pri_index
1210     or
1211       do { # get another row, stash it, drop out if different PK
1212
1213         @copy = $self->cursor->next;
1214         $self->{stashed_row} = \@copy;
1215
1216         # last thing in do block, counts as true if anything doesn't match
1217
1218         # check xor defined first for NULL vs. NOT NULL then if one is
1219         # defined the other must be so check string equality
1220
1221         grep {
1222           (defined $pri_vals{$_} ^ defined $copy[$_])
1223           || (defined $pri_vals{$_} && ($pri_vals{$_} ne $copy[$_]))
1224         } @pri_index;
1225       }
1226   );
1227
1228   my $alias = $self->{attrs}{alias};
1229   my $info = [];
1230
1231   my %collapse_pos;
1232
1233   my @const_keys;
1234
1235   foreach my $const (@const_rows) {
1236     scalar @const_keys or do {
1237       @const_keys = sort { length($a) <=> length($b) } keys %$const;
1238     };
1239     foreach my $key (@const_keys) {
1240       if (length $key) {
1241         my $target = $info;
1242         my @parts = split(/\./, $key);
1243         my $cur = '';
1244         my $data = $const->{$key};
1245         foreach my $p (@parts) {
1246           $target = $target->[1]->{$p} ||= [];
1247           $cur .= ".${p}";
1248           if ($cur eq ".${key}" && (my @ckey = @{$collapse{$cur}||[]})) {
1249             # collapsing at this point and on final part
1250             my $pos = $collapse_pos{$cur};
1251             CK: foreach my $ck (@ckey) {
1252               if (!defined $pos->{$ck} || $pos->{$ck} ne $data->{$ck}) {
1253                 $collapse_pos{$cur} = $data;
1254                 delete @collapse_pos{ # clear all positioning for sub-entries
1255                   grep { m/^\Q${cur}.\E/ } keys %collapse_pos
1256                 };
1257                 push(@$target, []);
1258                 last CK;
1259               }
1260             }
1261           }
1262           if (exists $collapse{$cur}) {
1263             $target = $target->[-1];
1264           }
1265         }
1266         $target->[0] = $data;
1267       } else {
1268         $info->[0] = $const->{$key};
1269       }
1270     }
1271   }
1272
1273   return $info;
1274 }
1275
1276 =head2 result_source
1277
1278 =over 4
1279
1280 =item Arguments: $result_source?
1281
1282 =item Return Value: $result_source
1283
1284 =back
1285
1286 An accessor for the primary ResultSource object from which this ResultSet
1287 is derived.
1288
1289 =head2 result_class
1290
1291 =over 4
1292
1293 =item Arguments: $result_class?
1294
1295 =item Return Value: $result_class
1296
1297 =back
1298
1299 An accessor for the class to use when creating row objects. Defaults to
1300 C<< result_source->result_class >> - which in most cases is the name of the
1301 L<"table"|DBIx::Class::Manual::Glossary/"ResultSource"> class.
1302
1303 Note that changing the result_class will also remove any components
1304 that were originally loaded in the source class via
1305 L<DBIx::Class::ResultSource/load_components>. Any overloaded methods
1306 in the original source class will not run.
1307
1308 =cut
1309
1310 sub result_class {
1311   my ($self, $result_class) = @_;
1312   if ($result_class) {
1313     unless (ref $result_class) { # don't fire this for an object
1314       $self->ensure_class_loaded($result_class);
1315     }
1316     $self->_result_class($result_class);
1317     # THIS LINE WOULD BE A BUG - this accessor specifically exists to
1318     # permit the user to set result class on one result set only; it only
1319     # chains if provided to search()
1320     #$self->{attrs}{result_class} = $result_class if ref $self;
1321   }
1322   $self->_result_class;
1323 }
1324
1325 =head2 count
1326
1327 =over 4
1328
1329 =item Arguments: $cond, \%attrs??
1330
1331 =item Return Value: $count
1332
1333 =back
1334
1335 Performs an SQL C<COUNT> with the same query as the resultset was built
1336 with to find the number of elements. Passing arguments is equivalent to
1337 C<< $rs->search ($cond, \%attrs)->count >>
1338
1339 =cut
1340
1341 sub count {
1342   my $self = shift;
1343   return $self->search(@_)->count if @_ and defined $_[0];
1344   return scalar @{ $self->get_cache } if $self->get_cache;
1345
1346   my $attrs = $self->_resolved_attrs_copy;
1347
1348   # this is a little optimization - it is faster to do the limit
1349   # adjustments in software, instead of a subquery
1350   my $rows = delete $attrs->{rows};
1351   my $offset = delete $attrs->{offset};
1352
1353   my $crs;
1354   if ($self->_has_resolved_attr (qw/collapse group_by/)) {
1355     $crs = $self->_count_subq_rs ($attrs);
1356   }
1357   else {
1358     $crs = $self->_count_rs ($attrs);
1359   }
1360   my $count = $crs->next;
1361
1362   $count -= $offset if $offset;
1363   $count = $rows if $rows and $rows < $count;
1364   $count = 0 if ($count < 0);
1365
1366   return $count;
1367 }
1368
1369 =head2 count_rs
1370
1371 =over 4
1372
1373 =item Arguments: $cond, \%attrs??
1374
1375 =item Return Value: $count_rs
1376
1377 =back
1378
1379 Same as L</count> but returns a L<DBIx::Class::ResultSetColumn> object.
1380 This can be very handy for subqueries:
1381
1382   ->search( { amount => $some_rs->count_rs->as_query } )
1383
1384 As with regular resultsets the SQL query will be executed only after
1385 the resultset is accessed via L</next> or L</all>. That would return
1386 the same single value obtainable via L</count>.
1387
1388 =cut
1389
1390 sub count_rs {
1391   my $self = shift;
1392   return $self->search(@_)->count_rs if @_;
1393
1394   # this may look like a lack of abstraction (count() does about the same)
1395   # but in fact an _rs *must* use a subquery for the limits, as the
1396   # software based limiting can not be ported if this $rs is to be used
1397   # in a subquery itself (i.e. ->as_query)
1398   if ($self->_has_resolved_attr (qw/collapse group_by offset rows/)) {
1399     return $self->_count_subq_rs;
1400   }
1401   else {
1402     return $self->_count_rs;
1403   }
1404 }
1405
1406 #
1407 # returns a ResultSetColumn object tied to the count query
1408 #
1409 sub _count_rs {
1410   my ($self, $attrs) = @_;
1411
1412   my $rsrc = $self->result_source;
1413   $attrs ||= $self->_resolved_attrs;
1414
1415   my $tmp_attrs = { %$attrs };
1416   # take off any limits, record_filter is cdbi, and no point of ordering nor locking a count
1417   delete @{$tmp_attrs}{qw/rows offset order_by record_filter for/};
1418
1419   # overwrite the selector (supplied by the storage)
1420   $tmp_attrs->{select} = $rsrc->storage->_count_select ($rsrc, $attrs);
1421   $tmp_attrs->{as} = 'count';
1422   delete @{$tmp_attrs}{qw/columns _trailing_select/};
1423
1424   my $tmp_rs = $rsrc->resultset_class->new($rsrc, $tmp_attrs)->get_column ('count');
1425
1426   return $tmp_rs;
1427 }
1428
1429 #
1430 # same as above but uses a subquery
1431 #
1432 sub _count_subq_rs {
1433   my ($self, $attrs) = @_;
1434
1435   my $rsrc = $self->result_source;
1436   $attrs ||= $self->_resolved_attrs;
1437
1438   my $sub_attrs = { %$attrs };
1439   # extra selectors do not go in the subquery and there is no point of ordering it, nor locking it
1440   delete @{$sub_attrs}{qw/collapse columns as select _prefetch_selector_range _trailing_select order_by for/};
1441
1442   # if we multi-prefetch we group_by primary keys only as this is what we would
1443   # get out of the rs via ->next/->all. We *DO WANT* to clobber old group_by regardless
1444   if ( keys %{$attrs->{collapse}}  ) {
1445     $sub_attrs->{group_by} = [ map { "$attrs->{alias}.$_" } ($rsrc->_pri_cols) ]
1446   }
1447
1448   # Calculate subquery selector
1449   if (my $g = $sub_attrs->{group_by}) {
1450
1451     my $sql_maker = $rsrc->storage->sql_maker;
1452
1453     # necessary as the group_by may refer to aliased functions
1454     my $sel_index;
1455     for my $sel (@{$attrs->{select}}) {
1456       $sel_index->{$sel->{-as}} = $sel
1457         if (ref $sel eq 'HASH' and $sel->{-as});
1458     }
1459
1460     # anything from the original select mentioned on the group-by needs to make it to the inner selector
1461     # also look for named aggregates referred in the having clause
1462     # having often contains scalarrefs - thus parse it out entirely
1463     my @parts = @$g;
1464     if ($attrs->{having}) {
1465       local $sql_maker->{having_bind};
1466       local $sql_maker->{quote_char} = $sql_maker->{quote_char};
1467       local $sql_maker->{name_sep} = $sql_maker->{name_sep};
1468       unless (defined $sql_maker->{quote_char} and length $sql_maker->{quote_char}) {
1469         $sql_maker->{quote_char} = [ "\x00", "\xFF" ];
1470         # if we don't unset it we screw up retarded but unfortunately working
1471         # 'MAX(foo.bar)' => { '>', 3 }
1472         $sql_maker->{name_sep} = '';
1473       }
1474
1475       my ($lquote, $rquote, $sep) = map { quotemeta $_ } ($sql_maker->_quote_chars, $sql_maker->name_sep);
1476
1477       my $sql = $sql_maker->_parse_rs_attrs ({ having => $attrs->{having} });
1478
1479       # search for both a proper quoted qualified string, for a naive unquoted scalarref
1480       # and if all fails for an utterly naive quoted scalar-with-function
1481       while ($sql =~ /
1482         $rquote $sep $lquote (.+?) $rquote
1483           |
1484         [\s,] \w+ \. (\w+) [\s,]
1485           |
1486         [\s,] $lquote (.+?) $rquote [\s,]
1487       /gx) {
1488         push @parts, ($1 || $2 || $3);  # one of them matched if we got here
1489       }
1490     }
1491
1492     for (@parts) {
1493       my $colpiece = $sel_index->{$_} || $_;
1494
1495       # unqualify join-based group_by's. Arcane but possible query
1496       # also horrible horrible hack to alias a column (not a func.)
1497       # (probably need to introduce SQLA syntax)
1498       if ($colpiece =~ /\./ && $colpiece !~ /^$attrs->{alias}\./) {
1499         my $as = $colpiece;
1500         $as =~ s/\./__/;
1501         $colpiece = \ sprintf ('%s AS %s', map { $sql_maker->_quote ($_) } ($colpiece, $as) );
1502       }
1503       push @{$sub_attrs->{select}}, $colpiece;
1504     }
1505   }
1506   else {
1507     my @pcols = map { "$attrs->{alias}.$_" } ($rsrc->primary_columns);
1508     $sub_attrs->{select} = @pcols ? \@pcols : [ 1 ];
1509   }
1510
1511   return $rsrc->resultset_class
1512                ->new ($rsrc, $sub_attrs)
1513                 ->as_subselect_rs
1514                  ->search ({}, { columns => { count => $rsrc->storage->_count_select ($rsrc, $attrs) } })
1515                   ->get_column ('count');
1516 }
1517
1518 sub _bool {
1519   return 1;
1520 }
1521
1522 =head2 count_literal
1523
1524 =over 4
1525
1526 =item Arguments: $sql_fragment, @bind_values
1527
1528 =item Return Value: $count
1529
1530 =back
1531
1532 Counts the results in a literal query. Equivalent to calling L</search_literal>
1533 with the passed arguments, then L</count>.
1534
1535 =cut
1536
1537 sub count_literal { shift->search_literal(@_)->count; }
1538
1539 =head2 all
1540
1541 =over 4
1542
1543 =item Arguments: none
1544
1545 =item Return Value: @objects
1546
1547 =back
1548
1549 Returns all elements in the resultset. Called implicitly if the resultset
1550 is returned in list context.
1551
1552 =cut
1553
1554 sub all {
1555   my $self = shift;
1556   if(@_) {
1557       $self->throw_exception("all() doesn't take any arguments, you probably wanted ->search(...)->all()");
1558   }
1559
1560   return @{ $self->get_cache } if $self->get_cache;
1561
1562   my @obj;
1563
1564   if (keys %{$self->_resolved_attrs->{collapse}}) {
1565     # Using $self->cursor->all is really just an optimisation.
1566     # If we're collapsing has_many prefetches it probably makes
1567     # very little difference, and this is cleaner than hacking
1568     # _construct_object to survive the approach
1569     $self->cursor->reset;
1570     my @row = $self->cursor->next;
1571     while (@row) {
1572       push(@obj, $self->_construct_object(@row));
1573       @row = (exists $self->{stashed_row}
1574                ? @{delete $self->{stashed_row}}
1575                : $self->cursor->next);
1576     }
1577   } else {
1578     @obj = map { $self->_construct_object(@$_) } $self->cursor->all;
1579   }
1580
1581   $self->set_cache(\@obj) if $self->{attrs}{cache};
1582
1583   return @obj;
1584 }
1585
1586 =head2 reset
1587
1588 =over 4
1589
1590 =item Arguments: none
1591
1592 =item Return Value: $self
1593
1594 =back
1595
1596 Resets the resultset's cursor, so you can iterate through the elements again.
1597 Implicitly resets the storage cursor, so a subsequent L</next> will trigger
1598 another query.
1599
1600 =cut
1601
1602 sub reset {
1603   my ($self) = @_;
1604   delete $self->{_attrs} if exists $self->{_attrs};
1605   $self->{all_cache_position} = 0;
1606   $self->cursor->reset;
1607   return $self;
1608 }
1609
1610 =head2 first
1611
1612 =over 4
1613
1614 =item Arguments: none
1615
1616 =item Return Value: $object | undef
1617
1618 =back
1619
1620 Resets the resultset and returns an object for the first result (or C<undef>
1621 if the resultset is empty).
1622
1623 =cut
1624
1625 sub first {
1626   return $_[0]->reset->next;
1627 }
1628
1629
1630 # _rs_update_delete
1631 #
1632 # Determines whether and what type of subquery is required for the $rs operation.
1633 # If grouping is necessary either supplies its own, or verifies the current one
1634 # After all is done delegates to the proper storage method.
1635
1636 sub _rs_update_delete {
1637   my ($self, $op, $values) = @_;
1638
1639   my $rsrc = $self->result_source;
1640
1641   # if a condition exists we need to strip all table qualifiers
1642   # if this is not possible we'll force a subquery below
1643   my $cond = $rsrc->schema->storage->_strip_cond_qualifiers ($self->{cond});
1644
1645   my $needs_group_by_subq = $self->_has_resolved_attr (qw/collapse group_by -join/);
1646   my $needs_subq = $needs_group_by_subq || (not defined $cond) || $self->_has_resolved_attr(qw/rows offset/);
1647
1648   if ($needs_group_by_subq or $needs_subq) {
1649
1650     # make a new $rs selecting only the PKs (that's all we really need)
1651     my $attrs = $self->_resolved_attrs_copy;
1652
1653
1654     delete $attrs->{$_} for qw/collapse _collapse_order_by select _prefetch_selector_range as/;
1655     $attrs->{columns} = [ map { "$attrs->{alias}.$_" } ($self->result_source->_pri_cols) ];
1656
1657     if ($needs_group_by_subq) {
1658       # make sure no group_by was supplied, or if there is one - make sure it matches
1659       # the columns compiled above perfectly. Anything else can not be sanely executed
1660       # on most databases so croak right then and there
1661
1662       if (my $g = $attrs->{group_by}) {
1663         my @current_group_by = map
1664           { $_ =~ /\./ ? $_ : "$attrs->{alias}.$_" }
1665           @$g
1666         ;
1667
1668         if (
1669           join ("\x00", sort @current_group_by)
1670             ne
1671           join ("\x00", sort @{$attrs->{columns}} )
1672         ) {
1673           $self->throw_exception (
1674             "You have just attempted a $op operation on a resultset which does group_by"
1675             . ' on columns other than the primary keys, while DBIC internally needs to retrieve'
1676             . ' the primary keys in a subselect. All sane RDBMS engines do not support this'
1677             . ' kind of queries. Please retry the operation with a modified group_by or'
1678             . ' without using one at all.'
1679           );
1680         }
1681       }
1682       else {
1683         $attrs->{group_by} = $attrs->{columns};
1684       }
1685     }
1686
1687     my $subrs = (ref $self)->new($rsrc, $attrs);
1688     return $self->result_source->storage->_subq_update_delete($subrs, $op, $values);
1689   }
1690   else {
1691     return $rsrc->storage->$op(
1692       $rsrc,
1693       $op eq 'update' ? $values : (),
1694       $cond,
1695     );
1696   }
1697 }
1698
1699 =head2 update
1700
1701 =over 4
1702
1703 =item Arguments: \%values
1704
1705 =item Return Value: $storage_rv
1706
1707 =back
1708
1709 Sets the specified columns in the resultset to the supplied values in a
1710 single query. Note that this will not run any accessor/set_column/update
1711 triggers, nor will it update any row object instances derived from this
1712 resultset (this includes the contents of the L<resultset cache|/set_cache>
1713 if any). See L</update_all> if you need to execute any on-update
1714 triggers or cascades defined either by you or a
1715 L<result component|DBIx::Class::Manual::Component/WHAT_IS_A_COMPONENT>.
1716
1717 The return value is a pass through of what the underlying
1718 storage backend returned, and may vary. See L<DBI/execute> for the most
1719 common case.
1720
1721 =head3 CAVEAT
1722
1723 Note that L</update> does not process/deflate any of the values passed in.
1724 This is unlike the corresponding L<DBIx::Class::Row/update>. The user must
1725 ensure manually that any value passed to this method will stringify to
1726 something the RDBMS knows how to deal with. A notable example is the
1727 handling of L<DateTime> objects, for more info see:
1728 L<DBIx::Class::Manual::Cookbook/Formatting_DateTime_objects_in_queries>.
1729
1730 =cut
1731
1732 sub update {
1733   my ($self, $values) = @_;
1734   $self->throw_exception('Values for update must be a hash')
1735     unless ref $values eq 'HASH';
1736
1737   return $self->_rs_update_delete ('update', $values);
1738 }
1739
1740 =head2 update_all
1741
1742 =over 4
1743
1744 =item Arguments: \%values
1745
1746 =item Return Value: 1
1747
1748 =back
1749
1750 Fetches all objects and updates them one at a time via
1751 L<DBIx::Class::Row/update>. Note that C<update_all> will run DBIC defined
1752 triggers, while L</update> will not.
1753
1754 =cut
1755
1756 sub update_all {
1757   my ($self, $values) = @_;
1758   $self->throw_exception('Values for update_all must be a hash')
1759     unless ref $values eq 'HASH';
1760
1761   my $guard = $self->result_source->schema->txn_scope_guard;
1762   $_->update($values) for $self->all;
1763   $guard->commit;
1764   return 1;
1765 }
1766
1767 =head2 delete
1768
1769 =over 4
1770
1771 =item Arguments: none
1772
1773 =item Return Value: $storage_rv
1774
1775 =back
1776
1777 Deletes the rows matching this resultset in a single query. Note that this
1778 will not run any delete triggers, nor will it alter the
1779 L<in_storage|DBIx::Class::Row/in_storage> status of any row object instances
1780 derived from this resultset (this includes the contents of the
1781 L<resultset cache|/set_cache> if any). See L</delete_all> if you need to
1782 execute any on-delete triggers or cascades defined either by you or a
1783 L<result component|DBIx::Class::Manual::Component/WHAT_IS_A_COMPONENT>.
1784
1785 The return value is a pass through of what the underlying storage backend
1786 returned, and may vary. See L<DBI/execute> for the most common case.
1787
1788 =cut
1789
1790 sub delete {
1791   my $self = shift;
1792   $self->throw_exception('delete does not accept any arguments')
1793     if @_;
1794
1795   return $self->_rs_update_delete ('delete');
1796 }
1797
1798 =head2 delete_all
1799
1800 =over 4
1801
1802 =item Arguments: none
1803
1804 =item Return Value: 1
1805
1806 =back
1807
1808 Fetches all objects and deletes them one at a time via
1809 L<DBIx::Class::Row/delete>. Note that C<delete_all> will run DBIC defined
1810 triggers, while L</delete> will not.
1811
1812 =cut
1813
1814 sub delete_all {
1815   my $self = shift;
1816   $self->throw_exception('delete_all does not accept any arguments')
1817     if @_;
1818
1819   my $guard = $self->result_source->schema->txn_scope_guard;
1820   $_->delete for $self->all;
1821   $guard->commit;
1822   return 1;
1823 }
1824
1825 =head2 populate
1826
1827 =over 4
1828
1829 =item Arguments: \@data;
1830
1831 =back
1832
1833 Accepts either an arrayref of hashrefs or alternatively an arrayref of arrayrefs.
1834 For the arrayref of hashrefs style each hashref should be a structure suitable
1835 forsubmitting to a $resultset->create(...) method.
1836
1837 In void context, C<insert_bulk> in L<DBIx::Class::Storage::DBI> is used
1838 to insert the data, as this is a faster method.
1839
1840 Otherwise, each set of data is inserted into the database using
1841 L<DBIx::Class::ResultSet/create>, and the resulting objects are
1842 accumulated into an array. The array itself, or an array reference
1843 is returned depending on scalar or list context.
1844
1845 Example:  Assuming an Artist Class that has many CDs Classes relating:
1846
1847   my $Artist_rs = $schema->resultset("Artist");
1848
1849   ## Void Context Example
1850   $Artist_rs->populate([
1851      { artistid => 4, name => 'Manufactured Crap', cds => [
1852         { title => 'My First CD', year => 2006 },
1853         { title => 'Yet More Tweeny-Pop crap', year => 2007 },
1854       ],
1855      },
1856      { artistid => 5, name => 'Angsty-Whiny Girl', cds => [
1857         { title => 'My parents sold me to a record company', year => 2005 },
1858         { title => 'Why Am I So Ugly?', year => 2006 },
1859         { title => 'I Got Surgery and am now Popular', year => 2007 }
1860       ],
1861      },
1862   ]);
1863
1864   ## Array Context Example
1865   my ($ArtistOne, $ArtistTwo, $ArtistThree) = $Artist_rs->populate([
1866     { name => "Artist One"},
1867     { name => "Artist Two"},
1868     { name => "Artist Three", cds=> [
1869     { title => "First CD", year => 2007},
1870     { title => "Second CD", year => 2008},
1871   ]}
1872   ]);
1873
1874   print $ArtistOne->name; ## response is 'Artist One'
1875   print $ArtistThree->cds->count ## reponse is '2'
1876
1877 For the arrayref of arrayrefs style,  the first element should be a list of the
1878 fieldsnames to which the remaining elements are rows being inserted.  For
1879 example:
1880
1881   $Arstist_rs->populate([
1882     [qw/artistid name/],
1883     [100, 'A Formally Unknown Singer'],
1884     [101, 'A singer that jumped the shark two albums ago'],
1885     [102, 'An actually cool singer'],
1886   ]);
1887
1888 Please note an important effect on your data when choosing between void and
1889 wantarray context. Since void context goes straight to C<insert_bulk> in
1890 L<DBIx::Class::Storage::DBI> this will skip any component that is overriding
1891 C<insert>.  So if you are using something like L<DBIx-Class-UUIDColumns> to
1892 create primary keys for you, you will find that your PKs are empty.  In this
1893 case you will have to use the wantarray context in order to create those
1894 values.
1895
1896 =cut
1897
1898 sub populate {
1899   my $self = shift;
1900
1901   # cruft placed in standalone method
1902   my $data = $self->_normalize_populate_args(@_);
1903
1904   if(defined wantarray) {
1905     my @created;
1906     foreach my $item (@$data) {
1907       push(@created, $self->create($item));
1908     }
1909     return wantarray ? @created : \@created;
1910   }
1911   else {
1912     my $first = $data->[0];
1913
1914     # if a column is a registered relationship, and is a non-blessed hash/array, consider
1915     # it relationship data
1916     my (@rels, @columns);
1917     my $rsrc = $self->result_source;
1918     my $rels = { map { $_ => $rsrc->relationship_info($_) } $rsrc->relationships };
1919     for (keys %$first) {
1920       my $ref = ref $first->{$_};
1921       $rels->{$_} && ($ref eq 'ARRAY' or $ref eq 'HASH')
1922         ? push @rels, $_
1923         : push @columns, $_
1924       ;
1925     }
1926
1927     my @pks = $rsrc->primary_columns;
1928
1929     ## do the belongs_to relationships
1930     foreach my $index (0..$#$data) {
1931
1932       # delegate to create() for any dataset without primary keys with specified relationships
1933       if (grep { !defined $data->[$index]->{$_} } @pks ) {
1934         for my $r (@rels) {
1935           if (grep { ref $data->[$index]{$r} eq $_ } qw/HASH ARRAY/) {  # a related set must be a HASH or AoH
1936             my @ret = $self->populate($data);
1937             return;
1938           }
1939         }
1940       }
1941
1942       foreach my $rel (@rels) {
1943         next unless ref $data->[$index]->{$rel} eq "HASH";
1944         my $result = $self->related_resultset($rel)->create($data->[$index]->{$rel});
1945         my ($reverse_relname, $reverse_relinfo) = %{$rsrc->reverse_relationship_info($rel)};
1946         my $related = $result->result_source->_resolve_condition(
1947           $reverse_relinfo->{cond},
1948           $self,
1949           $result,
1950         );
1951
1952         delete $data->[$index]->{$rel};
1953         $data->[$index] = {%{$data->[$index]}, %$related};
1954
1955         push @columns, keys %$related if $index == 0;
1956       }
1957     }
1958
1959     ## inherit the data locked in the conditions of the resultset
1960     my ($rs_data) = $self->_merge_with_rscond({});
1961     delete @{$rs_data}{@columns};
1962     my @inherit_cols = keys %$rs_data;
1963     my @inherit_data = values %$rs_data;
1964
1965     ## do bulk insert on current row
1966     $rsrc->storage->insert_bulk(
1967       $rsrc,
1968       [@columns, @inherit_cols],
1969       [ map { [ @$_{@columns}, @inherit_data ] } @$data ],
1970     );
1971
1972     ## do the has_many relationships
1973     foreach my $item (@$data) {
1974
1975       my $main_row;
1976
1977       foreach my $rel (@rels) {
1978         next unless ref $item->{$rel} eq "ARRAY" && @{ $item->{$rel} };
1979
1980         $main_row ||= $self->new_result({map { $_ => $item->{$_} } @pks});
1981
1982         my $child = $main_row->$rel;
1983
1984         my $related = $child->result_source->_resolve_condition(
1985           $rels->{$rel}{cond},
1986           $child,
1987           $main_row,
1988         );
1989
1990         my @rows_to_add = ref $item->{$rel} eq 'ARRAY' ? @{$item->{$rel}} : ($item->{$rel});
1991         my @populate = map { {%$_, %$related} } @rows_to_add;
1992
1993         $child->populate( \@populate );
1994       }
1995     }
1996   }
1997 }
1998
1999
2000 # populate() argumnets went over several incarnations
2001 # What we ultimately support is AoH
2002 sub _normalize_populate_args {
2003   my ($self, $arg) = @_;
2004
2005   if (ref $arg eq 'ARRAY') {
2006     if (ref $arg->[0] eq 'HASH') {
2007       return $arg;
2008     }
2009     elsif (ref $arg->[0] eq 'ARRAY') {
2010       my @ret;
2011       my @colnames = @{$arg->[0]};
2012       foreach my $values (@{$arg}[1 .. $#$arg]) {
2013         push @ret, { map { $colnames[$_] => $values->[$_] } (0 .. $#colnames) };
2014       }
2015       return \@ret;
2016     }
2017   }
2018
2019   $self->throw_exception('Populate expects an arrayref of hashrefs or arrayref of arrayrefs');
2020 }
2021
2022 =head2 pager
2023
2024 =over 4
2025
2026 =item Arguments: none
2027
2028 =item Return Value: $pager
2029
2030 =back
2031
2032 Return Value a L<Data::Page> object for the current resultset. Only makes
2033 sense for queries with a C<page> attribute.
2034
2035 To get the full count of entries for a paged resultset, call
2036 C<total_entries> on the L<Data::Page> object.
2037
2038 =cut
2039
2040 # make a wizard good for both a scalar and a hashref
2041 my $mk_lazy_count_wizard = sub {
2042   require Variable::Magic;
2043
2044   my $stash = { total_rs => shift };
2045   my $slot = shift; # only used by the hashref magic
2046
2047   my $magic = Variable::Magic::wizard (
2048     data => sub { $stash },
2049
2050     (!$slot)
2051     ? (
2052       # the scalar magic
2053       get => sub {
2054         # set value lazily, and dispell for good
2055         ${$_[0]} = $_[1]{total_rs}->count;
2056         Variable::Magic::dispell (${$_[0]}, $_[1]{magic_selfref});
2057         return 1;
2058       },
2059       set => sub {
2060         # an explicit set implies dispell as well
2061         # the unless() is to work around "fun and giggles" below
2062         Variable::Magic::dispell (${$_[0]}, $_[1]{magic_selfref})
2063           unless (caller(2))[3] eq 'DBIx::Class::ResultSet::pager';
2064         return 1;
2065       },
2066     )
2067     : (
2068       # the uvar magic
2069       fetch => sub {
2070         if ($_[2] eq $slot and !$_[1]{inactive}) {
2071           my $cnt = $_[1]{total_rs}->count;
2072           $_[0]->{$slot} = $cnt;
2073
2074           # attempting to dispell in a fetch handle (works in store), seems
2075           # to invariable segfault on 5.10, 5.12, 5.13 :(
2076           # so use an inactivator instead
2077           #Variable::Magic::dispell (%{$_[0]}, $_[1]{magic_selfref});
2078           $_[1]{inactive}++;
2079         }
2080         return 1;
2081       },
2082       store => sub {
2083         if (! $_[1]{inactive} and $_[2] eq $slot) {
2084           #Variable::Magic::dispell (%{$_[0]}, $_[1]{magic_selfref});
2085           $_[1]{inactive}++
2086             unless (caller(2))[3] eq 'DBIx::Class::ResultSet::pager';
2087         }
2088         return 1;
2089       },
2090     ),
2091   );
2092
2093   $stash->{magic_selfref} = $magic;
2094   weaken ($stash->{magic_selfref}); # this fails on 5.8.1
2095
2096   return $magic;
2097 };
2098
2099 # the tie class for 5.8.1
2100 {
2101   package # hide from pause
2102     DBIx::Class::__DBIC_LAZY_RS_COUNT__;
2103   use base qw/Tie::Hash/;
2104
2105   sub FIRSTKEY { my $dummy = scalar keys %{$_[0]{data}}; each %{$_[0]{data}} }
2106   sub NEXTKEY  { each %{$_[0]{data}} }
2107   sub EXISTS   { exists $_[0]{data}{$_[1]} }
2108   sub DELETE   { delete $_[0]{data}{$_[1]} }
2109   sub CLEAR    { %{$_[0]{data}} = () }
2110   sub SCALAR   { scalar %{$_[0]{data}} }
2111
2112   sub TIEHASH {
2113     $_[1]{data} = {%{$_[1]{selfref}}};
2114     %{$_[1]{selfref}} = ();
2115     Scalar::Util::weaken ($_[1]{selfref});
2116     return bless ($_[1], $_[0]);
2117   };
2118
2119   sub FETCH {
2120     if ($_[1] eq $_[0]{slot}) {
2121       my $cnt = $_[0]{data}{$_[1]} = $_[0]{total_rs}->count;
2122       untie %{$_[0]{selfref}};
2123       %{$_[0]{selfref}} = %{$_[0]{data}};
2124       return $cnt;
2125     }
2126     else {
2127       $_[0]{data}{$_[1]};
2128     }
2129   }
2130
2131   sub STORE {
2132     $_[0]{data}{$_[1]} = $_[2];
2133     if ($_[1] eq $_[0]{slot}) {
2134       untie %{$_[0]{selfref}};
2135       %{$_[0]{selfref}} = %{$_[0]{data}};
2136     }
2137     $_[2];
2138   }
2139 }
2140
2141 sub pager {
2142   my ($self) = @_;
2143
2144   return $self->{pager} if $self->{pager};
2145
2146   if ($self->get_cache) {
2147     $self->throw_exception ('Pagers on cached resultsets are not supported');
2148   }
2149
2150   my $attrs = $self->{attrs};
2151   if (!defined $attrs->{page}) {
2152     $self->throw_exception("Can't create pager for non-paged rs");
2153   }
2154   elsif ($attrs->{page} <= 0) {
2155     $self->throw_exception('Invalid page number (page-numbers are 1-based)');
2156   }
2157   $attrs->{rows} ||= 10;
2158
2159   # throw away the paging flags and re-run the count (possibly
2160   # with a subselect) to get the real total count
2161   my $count_attrs = { %$attrs };
2162   delete $count_attrs->{$_} for qw/rows offset page pager/;
2163   my $total_rs = (ref $self)->new($self->result_source, $count_attrs);
2164
2165
2166 ### the following may seem awkward and dirty, but it's a thought-experiment
2167 ### necessary for future development of DBIx::DS. Do *NOT* change this code
2168 ### before talking to ribasushi/mst
2169
2170   my $pager = Data::Page->new(
2171     0,  #start with an empty set
2172     $attrs->{rows},
2173     $self->{attrs}{page},
2174   );
2175
2176   my $data_slot = 'total_entries';
2177
2178   # Since we are interested in a cached value (once it's set - it's set), every
2179   # technique will detach from the magic-host once the time comes to fire the
2180   # ->count (or in the segfaulting case of >= 5.10 it will deactivate itself)
2181
2182   if ($] < 5.008003) {
2183     # 5.8.1 throws 'Modification of a read-only value attempted' when one tries
2184     # to weakref the magic container :(
2185     # tested on 5.8.1
2186     tie (%$pager, 'DBIx::Class::__DBIC_LAZY_RS_COUNT__',
2187       { slot => $data_slot, total_rs => $total_rs, selfref => $pager }
2188     );
2189   }
2190   elsif ($] < 5.010) {
2191     # We can use magic on the hash value slot. It's interesting that the magic is
2192     # attached to the hash-slot, and does *not* stop working once I do the dummy
2193     # assignments after the cast()
2194     # tested on 5.8.3 and 5.8.9
2195     my $magic = $mk_lazy_count_wizard->($total_rs);
2196     Variable::Magic::cast ( $pager->{$data_slot}, $magic );
2197
2198     # this is for fun and giggles
2199     $pager->{$data_slot} = -1;
2200     $pager->{$data_slot} = 0;
2201
2202     # this does not work for scalars, but works with
2203     # uvar magic below
2204     #my %vals = %$pager;
2205     #%$pager = ();
2206     #%{$pager} = %vals;
2207   }
2208   else {
2209     # And the uvar magic
2210     # works on 5.10.1, 5.12.1 and 5.13.4 in its current form,
2211     # however see the wizard maker for more notes
2212     my $magic = $mk_lazy_count_wizard->($total_rs, $data_slot);
2213     Variable::Magic::cast ( %$pager, $magic );
2214
2215     # still works
2216     $pager->{$data_slot} = -1;
2217     $pager->{$data_slot} = 0;
2218
2219     # this now works
2220     my %vals = %$pager;
2221     %$pager = ();
2222     %{$pager} = %vals;
2223   }
2224
2225   return $self->{pager} = $pager;
2226 }
2227
2228 =head2 page
2229
2230 =over 4
2231
2232 =item Arguments: $page_number
2233
2234 =item Return Value: $rs
2235
2236 =back
2237
2238 Returns a resultset for the $page_number page of the resultset on which page
2239 is called, where each page contains a number of rows equal to the 'rows'
2240 attribute set on the resultset (10 by default).
2241
2242 =cut
2243
2244 sub page {
2245   my ($self, $page) = @_;
2246   return (ref $self)->new($self->result_source, { %{$self->{attrs}}, page => $page });
2247 }
2248
2249 =head2 new_result
2250
2251 =over 4
2252
2253 =item Arguments: \%vals
2254
2255 =item Return Value: $rowobject
2256
2257 =back
2258
2259 Creates a new row object in the resultset's result class and returns
2260 it. The row is not inserted into the database at this point, call
2261 L<DBIx::Class::Row/insert> to do that. Calling L<DBIx::Class::Row/in_storage>
2262 will tell you whether the row object has been inserted or not.
2263
2264 Passes the hashref of input on to L<DBIx::Class::Row/new>.
2265
2266 =cut
2267
2268 sub new_result {
2269   my ($self, $values) = @_;
2270   $self->throw_exception( "new_result needs a hash" )
2271     unless (ref $values eq 'HASH');
2272
2273   my ($merged_cond, $cols_from_relations) = $self->_merge_with_rscond($values);
2274
2275   my %new = (
2276     %$merged_cond,
2277     @$cols_from_relations
2278       ? (-cols_from_relations => $cols_from_relations)
2279       : (),
2280     -result_source => $self->result_source, # DO NOT REMOVE THIS, REQUIRED
2281   );
2282
2283   return $self->result_class->new(\%new);
2284 }
2285
2286 # _merge_with_rscond
2287 #
2288 # Takes a simple hash of K/V data and returns its copy merged with the
2289 # condition already present on the resultset. Additionally returns an
2290 # arrayref of value/condition names, which were inferred from related
2291 # objects (this is needed for in-memory related objects)
2292 sub _merge_with_rscond {
2293   my ($self, $data) = @_;
2294
2295   my (%new_data, @cols_from_relations);
2296
2297   my $alias = $self->{attrs}{alias};
2298
2299   if (! defined $self->{cond}) {
2300     # just massage $data below
2301   }
2302   elsif ($self->{cond} eq $DBIx::Class::ResultSource::UNRESOLVABLE_CONDITION) {
2303     %new_data = %{ $self->{attrs}{related_objects} || {} };  # nothing might have been inserted yet
2304     @cols_from_relations = keys %new_data;
2305   }
2306   elsif (ref $self->{cond} ne 'HASH') {
2307     $self->throw_exception(
2308       "Can't abstract implicit construct, resultset condition not a hash"
2309     );
2310   }
2311   else {
2312     # precendence must be given to passed values over values inherited from
2313     # the cond, so the order here is important.
2314     my $collapsed_cond = $self->_collapse_cond($self->{cond});
2315     my %implied = %{$self->_remove_alias($collapsed_cond, $alias)};
2316
2317     while ( my($col, $value) = each %implied ) {
2318       my $vref = ref $value;
2319       if ($vref eq 'HASH' && keys(%$value) && (keys %$value)[0] eq '=') {
2320         $new_data{$col} = $value->{'='};
2321       }
2322       elsif( !$vref or $vref eq 'SCALAR' or blessed($value) ) {
2323         $new_data{$col} = $value;
2324       }
2325     }
2326   }
2327
2328   %new_data = (
2329     %new_data,
2330     %{ $self->_remove_alias($data, $alias) },
2331   );
2332
2333   return (\%new_data, \@cols_from_relations);
2334 }
2335
2336 # _has_resolved_attr
2337 #
2338 # determines if the resultset defines at least one
2339 # of the attributes supplied
2340 #
2341 # used to determine if a subquery is neccessary
2342 #
2343 # supports some virtual attributes:
2344 #   -join
2345 #     This will scan for any joins being present on the resultset.
2346 #     It is not a mere key-search but a deep inspection of {from}
2347 #
2348
2349 sub _has_resolved_attr {
2350   my ($self, @attr_names) = @_;
2351
2352   my $attrs = $self->_resolved_attrs;
2353
2354   my %extra_checks;
2355
2356   for my $n (@attr_names) {
2357     if (grep { $n eq $_ } (qw/-join/) ) {
2358       $extra_checks{$n}++;
2359       next;
2360     }
2361
2362     my $attr =  $attrs->{$n};
2363
2364     next if not defined $attr;
2365
2366     if (ref $attr eq 'HASH') {
2367       return 1 if keys %$attr;
2368     }
2369     elsif (ref $attr eq 'ARRAY') {
2370       return 1 if @$attr;
2371     }
2372     else {
2373       return 1 if $attr;
2374     }
2375   }
2376
2377   # a resolved join is expressed as a multi-level from
2378   return 1 if (
2379     $extra_checks{-join}
2380       and
2381     ref $attrs->{from} eq 'ARRAY'
2382       and
2383     @{$attrs->{from}} > 1
2384   );
2385
2386   return 0;
2387 }
2388
2389 # _collapse_cond
2390 #
2391 # Recursively collapse the condition.
2392
2393 sub _collapse_cond {
2394   my ($self, $cond, $collapsed) = @_;
2395
2396   $collapsed ||= {};
2397
2398   if (ref $cond eq 'ARRAY') {
2399     foreach my $subcond (@$cond) {
2400       next unless ref $subcond;  # -or
2401       $collapsed = $self->_collapse_cond($subcond, $collapsed);
2402     }
2403   }
2404   elsif (ref $cond eq 'HASH') {
2405     if (keys %$cond and (keys %$cond)[0] eq '-and') {
2406       foreach my $subcond (@{$cond->{-and}}) {
2407         $collapsed = $self->_collapse_cond($subcond, $collapsed);
2408       }
2409     }
2410     else {
2411       foreach my $col (keys %$cond) {
2412         my $value = $cond->{$col};
2413         $collapsed->{$col} = $value;
2414       }
2415     }
2416   }
2417
2418   return $collapsed;
2419 }
2420
2421 # _remove_alias
2422 #
2423 # Remove the specified alias from the specified query hash. A copy is made so
2424 # the original query is not modified.
2425
2426 sub _remove_alias {
2427   my ($self, $query, $alias) = @_;
2428
2429   my %orig = %{ $query || {} };
2430   my %unaliased;
2431
2432   foreach my $key (keys %orig) {
2433     if ($key !~ /\./) {
2434       $unaliased{$key} = $orig{$key};
2435       next;
2436     }
2437     $unaliased{$1} = $orig{$key}
2438       if $key =~ m/^(?:\Q$alias\E\.)?([^.]+)$/;
2439   }
2440
2441   return \%unaliased;
2442 }
2443
2444 =head2 as_query
2445
2446 =over 4
2447
2448 =item Arguments: none
2449
2450 =item Return Value: \[ $sql, @bind ]
2451
2452 =back
2453
2454 Returns the SQL query and bind vars associated with the invocant.
2455
2456 This is generally used as the RHS for a subquery.
2457
2458 =cut
2459
2460 sub as_query {
2461   my $self = shift;
2462
2463   my $attrs = $self->_resolved_attrs_copy;
2464
2465   # For future use:
2466   #
2467   # in list ctx:
2468   # my ($sql, \@bind, \%dbi_bind_attrs) = _select_args_to_query (...)
2469   # $sql also has no wrapping parenthesis in list ctx
2470   #
2471   my $sqlbind = $self->result_source->storage
2472     ->_select_args_to_query ($attrs->{from}, $attrs->{select}, $attrs->{where}, $attrs);
2473
2474   return $sqlbind;
2475 }
2476
2477 =head2 find_or_new
2478
2479 =over 4
2480
2481 =item Arguments: \%vals, \%attrs?
2482
2483 =item Return Value: $rowobject
2484
2485 =back
2486
2487   my $artist = $schema->resultset('Artist')->find_or_new(
2488     { artist => 'fred' }, { key => 'artists' });
2489
2490   $cd->cd_to_producer->find_or_new({ producer => $producer },
2491                                    { key => 'primary });
2492
2493 Find an existing record from this resultset using L</find>. if none exists,
2494 instantiate a new result object and return it. The object will not be saved
2495 into your storage until you call L<DBIx::Class::Row/insert> on it.
2496
2497 You most likely want this method when looking for existing rows using a unique
2498 constraint that is not the primary key, or looking for related rows.
2499
2500 If you want objects to be saved immediately, use L</find_or_create> instead.
2501
2502 B<Note>: Make sure to read the documentation of L</find> and understand the
2503 significance of the C<key> attribute, as its lack may skew your search, and
2504 subsequently result in spurious new objects.
2505
2506 B<Note>: Take care when using C<find_or_new> with a table having
2507 columns with default values that you intend to be automatically
2508 supplied by the database (e.g. an auto_increment primary key column).
2509 In normal usage, the value of such columns should NOT be included at
2510 all in the call to C<find_or_new>, even when set to C<undef>.
2511
2512 =cut
2513
2514 sub find_or_new {
2515   my $self     = shift;
2516   my $attrs    = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
2517   my $hash     = ref $_[0] eq 'HASH' ? shift : {@_};
2518   if (keys %$hash and my $row = $self->find($hash, $attrs) ) {
2519     return $row;
2520   }
2521   return $self->new_result($hash);
2522 }
2523
2524 =head2 create
2525
2526 =over 4
2527
2528 =item Arguments: \%vals
2529
2530 =item Return Value: a L<DBIx::Class::Row> $object
2531
2532 =back
2533
2534 Attempt to create a single new row or a row with multiple related rows
2535 in the table represented by the resultset (and related tables). This
2536 will not check for duplicate rows before inserting, use
2537 L</find_or_create> to do that.
2538
2539 To create one row for this resultset, pass a hashref of key/value
2540 pairs representing the columns of the table and the values you wish to
2541 store. If the appropriate relationships are set up, foreign key fields
2542 can also be passed an object representing the foreign row, and the
2543 value will be set to its primary key.
2544
2545 To create related objects, pass a hashref of related-object column values
2546 B<keyed on the relationship name>. If the relationship is of type C<multi>
2547 (L<DBIx::Class::Relationship/has_many>) - pass an arrayref of hashrefs.
2548 The process will correctly identify columns holding foreign keys, and will
2549 transparently populate them from the keys of the corresponding relation.
2550 This can be applied recursively, and will work correctly for a structure
2551 with an arbitrary depth and width, as long as the relationships actually
2552 exists and the correct column data has been supplied.
2553
2554
2555 Instead of hashrefs of plain related data (key/value pairs), you may
2556 also pass new or inserted objects. New objects (not inserted yet, see
2557 L</new>), will be inserted into their appropriate tables.
2558
2559 Effectively a shortcut for C<< ->new_result(\%vals)->insert >>.
2560
2561 Example of creating a new row.
2562
2563   $person_rs->create({
2564     name=>"Some Person",
2565     email=>"somebody@someplace.com"
2566   });
2567
2568 Example of creating a new row and also creating rows in a related C<has_many>
2569 or C<has_one> resultset.  Note Arrayref.
2570
2571   $artist_rs->create(
2572      { artistid => 4, name => 'Manufactured Crap', cds => [
2573         { title => 'My First CD', year => 2006 },
2574         { title => 'Yet More Tweeny-Pop crap', year => 2007 },
2575       ],
2576      },
2577   );
2578
2579 Example of creating a new row and also creating a row in a related
2580 C<belongs_to> resultset. Note Hashref.
2581
2582   $cd_rs->create({
2583     title=>"Music for Silly Walks",
2584     year=>2000,
2585     artist => {
2586       name=>"Silly Musician",
2587     }
2588   });
2589
2590 =over
2591
2592 =item WARNING
2593
2594 When subclassing ResultSet never attempt to override this method. Since
2595 it is a simple shortcut for C<< $self->new_result($attrs)->insert >>, a
2596 lot of the internals simply never call it, so your override will be
2597 bypassed more often than not. Override either L<new|DBIx::Class::Row/new>
2598 or L<insert|DBIx::Class::Row/insert> depending on how early in the
2599 L</create> process you need to intervene.
2600
2601 =back
2602
2603 =cut
2604
2605 sub create {
2606   my ($self, $attrs) = @_;
2607   $self->throw_exception( "create needs a hashref" )
2608     unless ref $attrs eq 'HASH';
2609   return $self->new_result($attrs)->insert;
2610 }
2611
2612 =head2 find_or_create
2613
2614 =over 4
2615
2616 =item Arguments: \%vals, \%attrs?
2617
2618 =item Return Value: $rowobject
2619
2620 =back
2621
2622   $cd->cd_to_producer->find_or_create({ producer => $producer },
2623                                       { key => 'primary' });
2624
2625 Tries to find a record based on its primary key or unique constraints; if none
2626 is found, creates one and returns that instead.
2627
2628   my $cd = $schema->resultset('CD')->find_or_create({
2629     cdid   => 5,
2630     artist => 'Massive Attack',
2631     title  => 'Mezzanine',
2632     year   => 2005,
2633   });
2634
2635 Also takes an optional C<key> attribute, to search by a specific key or unique
2636 constraint. For example:
2637
2638   my $cd = $schema->resultset('CD')->find_or_create(
2639     {
2640       artist => 'Massive Attack',
2641       title  => 'Mezzanine',
2642     },
2643     { key => 'cd_artist_title' }
2644   );
2645
2646 B<Note>: Make sure to read the documentation of L</find> and understand the
2647 significance of the C<key> attribute, as its lack may skew your search, and
2648 subsequently result in spurious row creation.
2649
2650 B<Note>: Because find_or_create() reads from the database and then
2651 possibly inserts based on the result, this method is subject to a race
2652 condition. Another process could create a record in the table after
2653 the find has completed and before the create has started. To avoid
2654 this problem, use find_or_create() inside a transaction.
2655
2656 B<Note>: Take care when using C<find_or_create> with a table having
2657 columns with default values that you intend to be automatically
2658 supplied by the database (e.g. an auto_increment primary key column).
2659 In normal usage, the value of such columns should NOT be included at
2660 all in the call to C<find_or_create>, even when set to C<undef>.
2661
2662 See also L</find> and L</update_or_create>. For information on how to declare
2663 unique constraints, see L<DBIx::Class::ResultSource/add_unique_constraint>.
2664
2665 =cut
2666
2667 sub find_or_create {
2668   my $self     = shift;
2669   my $attrs    = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
2670   my $hash     = ref $_[0] eq 'HASH' ? shift : {@_};
2671   if (keys %$hash and my $row = $self->find($hash, $attrs) ) {
2672     return $row;
2673   }
2674   return $self->create($hash);
2675 }
2676
2677 =head2 update_or_create
2678
2679 =over 4
2680
2681 =item Arguments: \%col_values, { key => $unique_constraint }?
2682
2683 =item Return Value: $row_object
2684
2685 =back
2686
2687   $resultset->update_or_create({ col => $val, ... });
2688
2689 Like L</find_or_create>, but if a row is found it is immediately updated via
2690 C<< $found_row->update (\%col_values) >>.
2691
2692
2693 Takes an optional C<key> attribute to search on a specific unique constraint.
2694 For example:
2695
2696   # In your application
2697   my $cd = $schema->resultset('CD')->update_or_create(
2698     {
2699       artist => 'Massive Attack',
2700       title  => 'Mezzanine',
2701       year   => 1998,
2702     },
2703     { key => 'cd_artist_title' }
2704   );
2705
2706   $cd->cd_to_producer->update_or_create({
2707     producer => $producer,
2708     name => 'harry',
2709   }, {
2710     key => 'primary',
2711   });
2712
2713 B<Note>: Make sure to read the documentation of L</find> and understand the
2714 significance of the C<key> attribute, as its lack may skew your search, and
2715 subsequently result in spurious row creation.
2716
2717 B<Note>: Take care when using C<update_or_create> with a table having
2718 columns with default values that you intend to be automatically
2719 supplied by the database (e.g. an auto_increment primary key column).
2720 In normal usage, the value of such columns should NOT be included at
2721 all in the call to C<update_or_create>, even when set to C<undef>.
2722
2723 See also L</find> and L</find_or_create>. For information on how to declare
2724 unique constraints, see L<DBIx::Class::ResultSource/add_unique_constraint>.
2725
2726 =cut
2727
2728 sub update_or_create {
2729   my $self = shift;
2730   my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
2731   my $cond = ref $_[0] eq 'HASH' ? shift : {@_};
2732
2733   my $row = $self->find($cond, $attrs);
2734   if (defined $row) {
2735     $row->update($cond);
2736     return $row;
2737   }
2738
2739   return $self->create($cond);
2740 }
2741
2742 =head2 update_or_new
2743
2744 =over 4
2745
2746 =item Arguments: \%col_values, { key => $unique_constraint }?
2747
2748 =item Return Value: $rowobject
2749
2750 =back
2751
2752   $resultset->update_or_new({ col => $val, ... });
2753
2754 Like L</find_or_new> but if a row is found it is immediately updated via
2755 C<< $found_row->update (\%col_values) >>.
2756
2757 For example:
2758
2759   # In your application
2760   my $cd = $schema->resultset('CD')->update_or_new(
2761     {
2762       artist => 'Massive Attack',
2763       title  => 'Mezzanine',
2764       year   => 1998,
2765     },
2766     { key => 'cd_artist_title' }
2767   );
2768
2769   if ($cd->in_storage) {
2770       # the cd was updated
2771   }
2772   else {
2773       # the cd is not yet in the database, let's insert it
2774       $cd->insert;
2775   }
2776
2777 B<Note>: Make sure to read the documentation of L</find> and understand the
2778 significance of the C<key> attribute, as its lack may skew your search, and
2779 subsequently result in spurious new objects.
2780
2781 B<Note>: Take care when using C<update_or_new> with a table having
2782 columns with default values that you intend to be automatically
2783 supplied by the database (e.g. an auto_increment primary key column).
2784 In normal usage, the value of such columns should NOT be included at
2785 all in the call to C<update_or_new>, even when set to C<undef>.
2786
2787 See also L</find>, L</find_or_create> and L</find_or_new>.
2788
2789 =cut
2790
2791 sub update_or_new {
2792     my $self  = shift;
2793     my $attrs = ( @_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {} );
2794     my $cond  = ref $_[0] eq 'HASH' ? shift : {@_};
2795
2796     my $row = $self->find( $cond, $attrs );
2797     if ( defined $row ) {
2798         $row->update($cond);
2799         return $row;
2800     }
2801
2802     return $self->new_result($cond);
2803 }
2804
2805 =head2 get_cache
2806
2807 =over 4
2808
2809 =item Arguments: none
2810
2811 =item Return Value: \@cache_objects | undef
2812
2813 =back
2814
2815 Gets the contents of the cache for the resultset, if the cache is set.
2816
2817 The cache is populated either by using the L</prefetch> attribute to
2818 L</search> or by calling L</set_cache>.
2819
2820 =cut
2821
2822 sub get_cache {
2823   shift->{all_cache};
2824 }
2825
2826 =head2 set_cache
2827
2828 =over 4
2829
2830 =item Arguments: \@cache_objects
2831
2832 =item Return Value: \@cache_objects
2833
2834 =back
2835
2836 Sets the contents of the cache for the resultset. Expects an arrayref
2837 of objects of the same class as those produced by the resultset. Note that
2838 if the cache is set the resultset will return the cached objects rather
2839 than re-querying the database even if the cache attr is not set.
2840
2841 The contents of the cache can also be populated by using the
2842 L</prefetch> attribute to L</search>.
2843
2844 =cut
2845
2846 sub set_cache {
2847   my ( $self, $data ) = @_;
2848   $self->throw_exception("set_cache requires an arrayref")
2849       if defined($data) && (ref $data ne 'ARRAY');
2850   $self->{all_cache} = $data;
2851 }
2852
2853 =head2 clear_cache
2854
2855 =over 4
2856
2857 =item Arguments: none
2858
2859 =item Return Value: undef
2860
2861 =back
2862
2863 Clears the cache for the resultset.
2864
2865 =cut
2866
2867 sub clear_cache {
2868   shift->set_cache(undef);
2869 }
2870
2871 =head2 is_paged
2872
2873 =over 4
2874
2875 =item Arguments: none
2876
2877 =item Return Value: true, if the resultset has been paginated
2878
2879 =back
2880
2881 =cut
2882
2883 sub is_paged {
2884   my ($self) = @_;
2885   return !!$self->{attrs}{page};
2886 }
2887
2888 =head2 is_ordered
2889
2890 =over 4
2891
2892 =item Arguments: none
2893
2894 =item Return Value: true, if the resultset has been ordered with C<order_by>.
2895
2896 =back
2897
2898 =cut
2899
2900 sub is_ordered {
2901   my ($self) = @_;
2902   return scalar $self->result_source->storage->_extract_order_criteria($self->{attrs}{order_by});
2903 }
2904
2905 =head2 related_resultset
2906
2907 =over 4
2908
2909 =item Arguments: $relationship_name
2910
2911 =item Return Value: $resultset
2912
2913 =back
2914
2915 Returns a related resultset for the supplied relationship name.
2916
2917   $artist_rs = $schema->resultset('CD')->related_resultset('Artist');
2918
2919 =cut
2920
2921 sub related_resultset {
2922   my ($self, $rel) = @_;
2923
2924   $self->{related_resultsets} ||= {};
2925   return $self->{related_resultsets}{$rel} ||= do {
2926     my $rsrc = $self->result_source;
2927     my $rel_info = $rsrc->relationship_info($rel);
2928
2929     $self->throw_exception(
2930       "search_related: result source '" . $rsrc->source_name .
2931         "' has no such relationship $rel")
2932       unless $rel_info;
2933
2934     my $attrs = $self->_chain_relationship($rel);
2935
2936     my $join_count = $attrs->{seen_join}{$rel};
2937
2938     my $alias = $self->result_source->storage
2939         ->relname_to_table_alias($rel, $join_count);
2940
2941     # since this is search_related, and we already slid the select window inwards
2942     # (the select/as attrs were deleted in the beginning), we need to flip all
2943     # left joins to inner, so we get the expected results
2944     # read the comment on top of the actual function to see what this does
2945     $attrs->{from} = $rsrc->schema->storage->_inner_join_to_node ($attrs->{from}, $alias);
2946
2947
2948     #XXX - temp fix for result_class bug. There likely is a more elegant fix -groditi
2949     delete @{$attrs}{qw(result_class alias)};
2950
2951     my $new_cache;
2952
2953     if (my $cache = $self->get_cache) {
2954       if ($cache->[0] && $cache->[0]->related_resultset($rel)->get_cache) {
2955         $new_cache = [ map { @{$_->related_resultset($rel)->get_cache} }
2956                         @$cache ];
2957       }
2958     }
2959
2960     my $rel_source = $rsrc->related_source($rel);
2961
2962     my $new = do {
2963
2964       # The reason we do this now instead of passing the alias to the
2965       # search_rs below is that if you wrap/overload resultset on the
2966       # source you need to know what alias it's -going- to have for things
2967       # to work sanely (e.g. RestrictWithObject wants to be able to add
2968       # extra query restrictions, and these may need to be $alias.)
2969
2970       my $rel_attrs = $rel_source->resultset_attributes;
2971       local $rel_attrs->{alias} = $alias;
2972
2973       $rel_source->resultset
2974                  ->search_rs(
2975                      undef, {
2976                        %$attrs,
2977                        where => $attrs->{where},
2978                    });
2979     };
2980     $new->set_cache($new_cache) if $new_cache;
2981     $new;
2982   };
2983 }
2984
2985 =head2 current_source_alias
2986
2987 =over 4
2988
2989 =item Arguments: none
2990
2991 =item Return Value: $source_alias
2992
2993 =back
2994
2995 Returns the current table alias for the result source this resultset is built
2996 on, that will be used in the SQL query. Usually it is C<me>.
2997
2998 Currently the source alias that refers to the result set returned by a
2999 L</search>/L</find> family method depends on how you got to the resultset: it's
3000 C<me> by default, but eg. L</search_related> aliases it to the related result
3001 source name (and keeps C<me> referring to the original result set). The long
3002 term goal is to make L<DBIx::Class> always alias the current resultset as C<me>
3003 (and make this method unnecessary).
3004
3005 Thus it's currently necessary to use this method in predefined queries (see
3006 L<DBIx::Class::Manual::Cookbook/Predefined searches>) when referring to the
3007 source alias of the current result set:
3008
3009   # in a result set class
3010   sub modified_by {
3011     my ($self, $user) = @_;
3012
3013     my $me = $self->current_source_alias;
3014
3015     return $self->search(
3016       "$me.modified" => $user->id,
3017     );
3018   }
3019
3020 =cut
3021
3022 sub current_source_alias {
3023   my ($self) = @_;
3024
3025   return ($self->{attrs} || {})->{alias} || 'me';
3026 }
3027
3028 =head2 as_subselect_rs
3029
3030 =over 4
3031
3032 =item Arguments: none
3033
3034 =item Return Value: $resultset
3035
3036 =back
3037
3038 Act as a barrier to SQL symbols.  The resultset provided will be made into a
3039 "virtual view" by including it as a subquery within the from clause.  From this
3040 point on, any joined tables are inaccessible to ->search on the resultset (as if
3041 it were simply where-filtered without joins).  For example:
3042
3043  my $rs = $schema->resultset('Bar')->search({'x.name' => 'abc'},{ join => 'x' });
3044
3045  # 'x' now pollutes the query namespace
3046
3047  # So the following works as expected
3048  my $ok_rs = $rs->search({'x.other' => 1});
3049
3050  # But this doesn't: instead of finding a 'Bar' related to two x rows (abc and
3051  # def) we look for one row with contradictory terms and join in another table
3052  # (aliased 'x_2') which we never use
3053  my $broken_rs = $rs->search({'x.name' => 'def'});
3054
3055  my $rs2 = $rs->as_subselect_rs;
3056
3057  # doesn't work - 'x' is no longer accessible in $rs2, having been sealed away
3058  my $not_joined_rs = $rs2->search({'x.other' => 1});
3059
3060  # works as expected: finds a 'table' row related to two x rows (abc and def)
3061  my $correctly_joined_rs = $rs2->search({'x.name' => 'def'});
3062
3063 Another example of when one might use this would be to select a subset of
3064 columns in a group by clause:
3065
3066  my $rs = $schema->resultset('Bar')->search(undef, {
3067    group_by => [qw{ id foo_id baz_id }],
3068  })->as_subselect_rs->search(undef, {
3069    columns => [qw{ id foo_id }]
3070  });
3071
3072 In the above example normally columns would have to be equal to the group by,
3073 but because we isolated the group by into a subselect the above works.
3074
3075 =cut
3076
3077 sub as_subselect_rs {
3078   my $self = shift;
3079
3080   my $attrs = $self->_resolved_attrs;
3081
3082   my $fresh_rs = (ref $self)->new (
3083     $self->result_source
3084   );
3085
3086   # these pieces will be locked in the subquery
3087   delete $fresh_rs->{cond};
3088   delete @{$fresh_rs->{attrs}}{qw/where bind/};
3089
3090   return $fresh_rs->search( {}, {
3091     from => [{
3092       $attrs->{alias} => $self->as_query,
3093       -alias         => $attrs->{alias},
3094       -source_handle => $self->result_source->handle,
3095     }],
3096     alias => $attrs->{alias},
3097   });
3098 }
3099
3100 # This code is called by search_related, and makes sure there
3101 # is clear separation between the joins before, during, and
3102 # after the relationship. This information is needed later
3103 # in order to properly resolve prefetch aliases (any alias
3104 # with a relation_chain_depth less than the depth of the
3105 # current prefetch is not considered)
3106 #
3107 # The increments happen twice per join. An even number means a
3108 # relationship specified via a search_related, whereas an odd
3109 # number indicates a join/prefetch added via attributes
3110 #
3111 # Also this code will wrap the current resultset (the one we
3112 # chain to) in a subselect IFF it contains limiting attributes
3113 sub _chain_relationship {
3114   my ($self, $rel) = @_;
3115   my $source = $self->result_source;
3116   my $attrs = { %{$self->{attrs}||{}} };
3117
3118   # we need to take the prefetch the attrs into account before we
3119   # ->_resolve_join as otherwise they get lost - captainL
3120   my $join = $self->_merge_joinpref_attr( $attrs->{join}, $attrs->{prefetch} );
3121
3122   delete @{$attrs}{qw/join prefetch collapse group_by distinct select as columns +select +as +columns/};
3123
3124   my $seen = { %{ (delete $attrs->{seen_join}) || {} } };
3125
3126   my $from;
3127   my @force_subq_attrs = qw/offset rows group_by having/;
3128
3129   if (
3130     ($attrs->{from} && ref $attrs->{from} ne 'ARRAY')
3131       ||
3132     $self->_has_resolved_attr (@force_subq_attrs)
3133   ) {
3134     # Nuke the prefetch (if any) before the new $rs attrs
3135     # are resolved (prefetch is useless - we are wrapping
3136     # a subquery anyway).
3137     my $rs_copy = $self->search;
3138     $rs_copy->{attrs}{join} = $self->_merge_joinpref_attr (
3139       $rs_copy->{attrs}{join},
3140       delete $rs_copy->{attrs}{prefetch},
3141     );
3142
3143     $from = [{
3144       -source_handle => $source->handle,
3145       -alias => $attrs->{alias},
3146       $attrs->{alias} => $rs_copy->as_query,
3147     }];
3148     delete @{$attrs}{@force_subq_attrs, qw/where bind/};
3149     $seen->{-relation_chain_depth} = 0;
3150   }
3151   elsif ($attrs->{from}) {  #shallow copy suffices
3152     $from = [ @{$attrs->{from}} ];
3153   }
3154   else {
3155     $from = [{
3156       -source_handle => $source->handle,
3157       -alias => $attrs->{alias},
3158       $attrs->{alias} => $source->from,
3159     }];
3160   }
3161
3162   my $jpath = ($seen->{-relation_chain_depth})
3163     ? $from->[-1][0]{-join_path}
3164     : [];
3165
3166   my @requested_joins = $source->_resolve_join(
3167     $join,
3168     $attrs->{alias},
3169     $seen,
3170     $jpath,
3171   );
3172
3173   push @$from, @requested_joins;
3174
3175   $seen->{-relation_chain_depth}++;
3176
3177   # if $self already had a join/prefetch specified on it, the requested
3178   # $rel might very well be already included. What we do in this case
3179   # is effectively a no-op (except that we bump up the chain_depth on
3180   # the join in question so we could tell it *is* the search_related)
3181   my $already_joined;
3182
3183   # we consider the last one thus reverse
3184   for my $j (reverse @requested_joins) {
3185     my ($last_j) = keys %{$j->[0]{-join_path}[-1]};
3186     if ($rel eq $last_j) {
3187       $j->[0]{-relation_chain_depth}++;
3188       $already_joined++;
3189       last;
3190     }
3191   }
3192
3193   unless ($already_joined) {
3194     push @$from, $source->_resolve_join(
3195       $rel,
3196       $attrs->{alias},
3197       $seen,
3198       $jpath,
3199     );
3200   }
3201
3202   $seen->{-relation_chain_depth}++;
3203
3204   return {%$attrs, from => $from, seen_join => $seen};
3205 }
3206
3207 # too many times we have to do $attrs = { %{$self->_resolved_attrs} }
3208 sub _resolved_attrs_copy {
3209   my $self = shift;
3210   return { %{$self->_resolved_attrs (@_)} };
3211 }
3212
3213 sub _resolved_attrs {
3214   my $self = shift;
3215   return $self->{_attrs} if $self->{_attrs};
3216
3217   my $attrs  = { %{ $self->{attrs} || {} } };
3218   my $source = $self->result_source;
3219   my $alias  = $attrs->{alias};
3220
3221   # one last pass of normalization
3222   $self->_normalize_selection($attrs);
3223
3224   # default selection list
3225   $attrs->{columns} = [ $source->columns ]
3226     unless List::Util::first { exists $attrs->{$_} } qw/columns cols select as _trailing_select/;
3227
3228   # merge selectors together
3229   for (qw/columns select as _trailing_select/) {
3230     $attrs->{$_} = $self->_merge_attr($attrs->{$_}, $attrs->{"+$_"})
3231       if $attrs->{$_} or $attrs->{"+$_"};
3232   }
3233
3234   # disassemble columns
3235   my (@sel, @as);
3236   if (my $cols = delete $attrs->{columns}) {
3237     for my $c (ref $cols eq 'ARRAY' ? @$cols : $cols) {
3238       if (ref $c eq 'HASH') {
3239         for my $as (keys %$c) {
3240           push @sel, $c->{$as};
3241           push @as, $as;
3242         }
3243       }
3244       else {
3245         push @sel, $c;
3246         push @as, $c;
3247       }
3248     }
3249   }
3250
3251   # when trying to weed off duplicates later do not go past this point -
3252   # everything added from here on is unbalanced "anyone's guess" stuff
3253   my $dedup_stop_idx = $#as;
3254
3255   push @as, @{ ref $attrs->{as} eq 'ARRAY' ? $attrs->{as} : [ $attrs->{as} ] }
3256     if $attrs->{as};
3257   push @sel, @{ ref $attrs->{select} eq 'ARRAY' ? $attrs->{select} : [ $attrs->{select} ] }
3258     if $attrs->{select};
3259
3260   # assume all unqualified selectors to apply to the current alias (legacy stuff)
3261   for (@sel) {
3262     $_ = (ref $_ or $_ =~ /\./) ? $_ : "$alias.$_";
3263   }
3264
3265   # disqualify all $alias.col as-bits (collapser mandated)
3266   for (@as) {
3267     $_ = ($_ =~ /^\Q$alias.\E(.+)$/) ? $1 : $_;
3268   }
3269
3270   # de-duplicate the result (remove *identical* select/as pairs)
3271   # and also die on duplicate {as} pointing to different {select}s
3272   # not using a c-style for as the condition is prone to shrinkage
3273   my $seen;
3274   my $i = 0;
3275   while ($i <= $dedup_stop_idx) {
3276     if ($seen->{"$sel[$i] \x00\x00 $as[$i]"}++) {
3277       splice @sel, $i, 1;
3278       splice @as, $i, 1;
3279       $dedup_stop_idx--;
3280     }
3281     elsif ($seen->{$as[$i]}++) {
3282       $self->throw_exception(
3283         "inflate_result() alias '$as[$i]' specified twice with different SQL-side {select}-ors"
3284       );
3285     }
3286     else {
3287       $i++;
3288     }
3289   }
3290
3291   $attrs->{select} = \@sel;
3292   $attrs->{as} = \@as;
3293
3294   $attrs->{from} ||= [{
3295     -source_handle => $source->handle,
3296     -alias => $self->{attrs}{alias},
3297     $self->{attrs}{alias} => $source->from,
3298   }];
3299
3300   if ( $attrs->{join} || $attrs->{prefetch} ) {
3301
3302     $self->throw_exception ('join/prefetch can not be used with a custom {from}')
3303       if ref $attrs->{from} ne 'ARRAY';
3304
3305     my $join = (delete $attrs->{join}) || {};
3306
3307     if ( defined $attrs->{prefetch} ) {
3308       $join = $self->_merge_joinpref_attr( $join, $attrs->{prefetch} );
3309     }
3310
3311     $attrs->{from} =    # have to copy here to avoid corrupting the original
3312       [
3313         @{ $attrs->{from} },
3314         $source->_resolve_join(
3315           $join,
3316           $alias,
3317           { %{ $attrs->{seen_join} || {} } },
3318           ( $attrs->{seen_join} && keys %{$attrs->{seen_join}})
3319             ? $attrs->{from}[-1][0]{-join_path}
3320             : []
3321           ,
3322         )
3323       ];
3324   }
3325
3326   if ( defined $attrs->{order_by} ) {
3327     $attrs->{order_by} = (
3328       ref( $attrs->{order_by} ) eq 'ARRAY'
3329       ? [ @{ $attrs->{order_by} } ]
3330       : [ $attrs->{order_by} || () ]
3331     );
3332   }
3333
3334   if ($attrs->{group_by} and ref $attrs->{group_by} ne 'ARRAY') {
3335     $attrs->{group_by} = [ $attrs->{group_by} ];
3336   }
3337
3338   # generate the distinct induced group_by early, as prefetch will be carried via a
3339   # subquery (since a group_by is present)
3340   if (delete $attrs->{distinct}) {
3341     if ($attrs->{group_by}) {
3342       carp ("Useless use of distinct on a grouped resultset ('distinct' is ignored when a 'group_by' is present)");
3343     }
3344     else {
3345       # distinct affects only the main selection part, not what prefetch may
3346       # add below. However trailing is not yet a part of the selection as
3347       # prefetch must insert before it
3348       $attrs->{group_by} = $source->storage->_group_over_selection (
3349         $attrs->{from},
3350         [ @{$attrs->{select}||[]}, @{$attrs->{_trailing_select}||[]} ],
3351         $attrs->{order_by},
3352       );
3353     }
3354   }
3355
3356   $attrs->{collapse} ||= {};
3357   if ($attrs->{prefetch}) {
3358     my $prefetch = $self->_merge_joinpref_attr( {}, delete $attrs->{prefetch} );
3359
3360     my $prefetch_ordering = [];
3361
3362     # this is a separate structure (we don't look in {from} directly)
3363     # as the resolver needs to shift things off the lists to work
3364     # properly (identical-prefetches on different branches)
3365     my $join_map = {};
3366     if (ref $attrs->{from} eq 'ARRAY') {
3367
3368       my $start_depth = $attrs->{seen_join}{-relation_chain_depth} || 0;
3369
3370       for my $j ( @{$attrs->{from}}[1 .. $#{$attrs->{from}} ] ) {
3371         next unless $j->[0]{-alias};
3372         next unless $j->[0]{-join_path};
3373         next if ($j->[0]{-relation_chain_depth} || 0) < $start_depth;
3374
3375         my @jpath = map { keys %$_ } @{$j->[0]{-join_path}};
3376
3377         my $p = $join_map;
3378         $p = $p->{$_} ||= {} for @jpath[ ($start_depth/2) .. $#jpath]; #only even depths are actual jpath boundaries
3379         push @{$p->{-join_aliases} }, $j->[0]{-alias};
3380       }
3381     }
3382
3383     my @prefetch =
3384       $source->_resolve_prefetch( $prefetch, $alias, $join_map, $prefetch_ordering, $attrs->{collapse} );
3385
3386     # we need to somehow mark which columns came from prefetch
3387     if (@prefetch) {
3388       my $sel_end = $#{$attrs->{select}};
3389       $attrs->{_prefetch_selector_range} = [ $sel_end + 1, $sel_end + @prefetch ];
3390     }
3391
3392     push @{ $attrs->{select} }, (map { $_->[0] } @prefetch);
3393     push @{ $attrs->{as} }, (map { $_->[1] } @prefetch);
3394
3395     push( @{$attrs->{order_by}}, @$prefetch_ordering );
3396     $attrs->{_collapse_order_by} = \@$prefetch_ordering;
3397   }
3398
3399
3400   push @{ $attrs->{select} }, @{$attrs->{_trailing_select}}
3401     if $attrs->{_trailing_select};
3402
3403   # if both page and offset are specified, produce a combined offset
3404   # even though it doesn't make much sense, this is what pre 081xx has
3405   # been doing
3406   if (my $page = delete $attrs->{page}) {
3407     $attrs->{offset} =
3408       ($attrs->{rows} * ($page - 1))
3409             +
3410       ($attrs->{offset} || 0)
3411     ;
3412   }
3413
3414   return $self->{_attrs} = $attrs;
3415 }
3416
3417 sub _rollout_attr {
3418   my ($self, $attr) = @_;
3419
3420   if (ref $attr eq 'HASH') {
3421     return $self->_rollout_hash($attr);
3422   } elsif (ref $attr eq 'ARRAY') {
3423     return $self->_rollout_array($attr);
3424   } else {
3425     return [$attr];
3426   }
3427 }
3428
3429 sub _rollout_array {
3430   my ($self, $attr) = @_;
3431
3432   my @rolled_array;
3433   foreach my $element (@{$attr}) {
3434     if (ref $element eq 'HASH') {
3435       push( @rolled_array, @{ $self->_rollout_hash( $element ) } );
3436     } elsif (ref $element eq 'ARRAY') {
3437       #  XXX - should probably recurse here
3438       push( @rolled_array, @{$self->_rollout_array($element)} );
3439     } else {
3440       push( @rolled_array, $element );
3441     }
3442   }
3443   return \@rolled_array;
3444 }
3445
3446 sub _rollout_hash {
3447   my ($self, $attr) = @_;
3448
3449   my @rolled_array;
3450   foreach my $key (keys %{$attr}) {
3451     push( @rolled_array, { $key => $attr->{$key} } );
3452   }
3453   return \@rolled_array;
3454 }
3455
3456 sub _calculate_score {
3457   my ($self, $a, $b) = @_;
3458
3459   if (defined $a xor defined $b) {
3460     return 0;
3461   }
3462   elsif (not defined $a) {
3463     return 1;
3464   }
3465
3466   if (ref $b eq 'HASH') {
3467     my ($b_key) = keys %{$b};
3468     if (ref $a eq 'HASH') {
3469       my ($a_key) = keys %{$a};
3470       if ($a_key eq $b_key) {
3471         return (1 + $self->_calculate_score( $a->{$a_key}, $b->{$b_key} ));
3472       } else {
3473         return 0;
3474       }
3475     } else {
3476       return ($a eq $b_key) ? 1 : 0;
3477     }
3478   } else {
3479     if (ref $a eq 'HASH') {
3480       my ($a_key) = keys %{$a};
3481       return ($b eq $a_key) ? 1 : 0;
3482     } else {
3483       return ($b eq $a) ? 1 : 0;
3484     }
3485   }
3486 }
3487
3488 sub _merge_joinpref_attr {
3489   my ($self, $orig, $import) = @_;
3490
3491   return $import unless defined($orig);
3492   return $orig unless defined($import);
3493
3494   $orig = $self->_rollout_attr($orig);
3495   $import = $self->_rollout_attr($import);
3496
3497   my $seen_keys;
3498   foreach my $import_element ( @{$import} ) {
3499     # find best candidate from $orig to merge $b_element into
3500     my $best_candidate = { position => undef, score => 0 }; my $position = 0;
3501     foreach my $orig_element ( @{$orig} ) {
3502       my $score = $self->_calculate_score( $orig_element, $import_element );
3503       if ($score > $best_candidate->{score}) {
3504         $best_candidate->{position} = $position;
3505         $best_candidate->{score} = $score;
3506       }
3507       $position++;
3508     }
3509     my ($import_key) = ( ref $import_element eq 'HASH' ) ? keys %{$import_element} : ($import_element);
3510
3511     if ($best_candidate->{score} == 0 || exists $seen_keys->{$import_key}) {
3512       push( @{$orig}, $import_element );
3513     } else {
3514       my $orig_best = $orig->[$best_candidate->{position}];
3515       # merge orig_best and b_element together and replace original with merged
3516       if (ref $orig_best ne 'HASH') {
3517         $orig->[$best_candidate->{position}] = $import_element;
3518       } elsif (ref $import_element eq 'HASH') {
3519         my ($key) = keys %{$orig_best};
3520         $orig->[$best_candidate->{position}] = { $key => $self->_merge_joinpref_attr($orig_best->{$key}, $import_element->{$key}) };
3521       }
3522     }
3523     $seen_keys->{$import_key} = 1; # don't merge the same key twice
3524   }
3525
3526   return $orig;
3527 }
3528
3529 {
3530   my $hm;
3531
3532   sub _merge_attr {
3533     $hm ||= do {
3534       my $hm = Hash::Merge->new;
3535
3536       $hm->specify_behavior({
3537         SCALAR => {
3538           SCALAR => sub {
3539             my ($defl, $defr) = map { defined $_ } (@_[0,1]);
3540
3541             if ($defl xor $defr) {
3542               return [ $defl ? $_[0] : $_[1] ];
3543             }
3544             elsif (! $defl) {
3545               return [];
3546             }
3547             elsif (__HM_DEDUP and $_[0] eq $_[1]) {
3548               return [ $_[0] ];
3549             }
3550             else {
3551               return [$_[0], $_[1]];
3552             }
3553           },
3554           ARRAY => sub {
3555             return $_[1] if !defined $_[0];
3556             return $_[1] if __HM_DEDUP and List::Util::first { $_ eq $_[0] } @{$_[1]};
3557             return [$_[0], @{$_[1]}]
3558           },
3559           HASH  => sub {
3560             return [] if !defined $_[0] and !keys %{$_[1]};
3561             return [ $_[1] ] if !defined $_[0];
3562             return [ $_[0] ] if !keys %{$_[1]};
3563             return [$_[0], $_[1]]
3564           },
3565         },
3566         ARRAY => {
3567           SCALAR => sub {
3568             return $_[0] if !defined $_[1];
3569             return $_[0] if __HM_DEDUP and List::Util::first { $_ eq $_[1] } @{$_[0]};
3570             return [@{$_[0]}, $_[1]]
3571           },
3572           ARRAY => sub {
3573             my @ret = @{$_[0]} or return $_[1];
3574             return [ @ret, @{$_[1]} ] unless __HM_DEDUP;
3575             my %idx = map { $_ => 1 } @ret;
3576             push @ret, grep { ! defined $idx{$_} } (@{$_[1]});
3577             \@ret;
3578           },
3579           HASH => sub {
3580             return [ $_[1] ] if ! @{$_[0]};
3581             return $_[0] if !keys %{$_[1]};
3582             return $_[0] if __HM_DEDUP and List::Util::first { $_ eq $_[1] } @{$_[0]};
3583             return [ @{$_[0]}, $_[1] ];
3584           },
3585         },
3586         HASH => {
3587           SCALAR => sub {
3588             return [] if !keys %{$_[0]} and !defined $_[1];
3589             return [ $_[0] ] if !defined $_[1];
3590             return [ $_[1] ] if !keys %{$_[0]};
3591             return [$_[0], $_[1]]
3592           },
3593           ARRAY => sub {
3594             return [] if !keys %{$_[0]} and !@{$_[1]};
3595             return [ $_[0] ] if !@{$_[1]};
3596             return $_[1] if !keys %{$_[0]};
3597             return $_[1] if __HM_DEDUP and List::Util::first { $_ eq $_[0] } @{$_[1]};
3598             return [ $_[0], @{$_[1]} ];
3599           },
3600           HASH => sub {
3601             return [] if !keys %{$_[0]} and !keys %{$_[1]};
3602             return [ $_[0] ] if !keys %{$_[1]};
3603             return [ $_[1] ] if !keys %{$_[0]};
3604             return [ $_[0] ] if $_[0] eq $_[1];
3605             return [ $_[0], $_[1] ];
3606           },
3607         }
3608       } => 'DBIC_RS_ATTR_MERGER');
3609       $hm;
3610     };
3611
3612     return $hm->merge ($_[1], $_[2]);
3613   }
3614 }
3615
3616 sub STORABLE_freeze {
3617   my ($self, $cloning) = @_;
3618   my $to_serialize = { %$self };
3619
3620   # A cursor in progress can't be serialized (and would make little sense anyway)
3621   delete $to_serialize->{cursor};
3622
3623   return nfreeze($to_serialize);
3624 }
3625
3626 # need this hook for symmetry
3627 sub STORABLE_thaw {
3628   my ($self, $cloning, $serialized) = @_;
3629
3630   %$self = %{ thaw($serialized) };
3631
3632   return $self;
3633 }
3634
3635
3636 =head2 throw_exception
3637
3638 See L<DBIx::Class::Schema/throw_exception> for details.
3639
3640 =cut
3641
3642 sub throw_exception {
3643   my $self=shift;
3644
3645   if (ref $self and my $rsrc = $self->result_source) {
3646     $rsrc->throw_exception(@_)
3647   }
3648   else {
3649     DBIx::Class::Exception->throw(@_);
3650   }
3651 }
3652
3653 # XXX: FIXME: Attributes docs need clearing up
3654
3655 =head1 ATTRIBUTES
3656
3657 Attributes are used to refine a ResultSet in various ways when
3658 searching for data. They can be passed to any method which takes an
3659 C<\%attrs> argument. See L</search>, L</search_rs>, L</find>,
3660 L</count>.
3661
3662 These are in no particular order:
3663
3664 =head2 order_by
3665
3666 =over 4
3667
3668 =item Value: ( $order_by | \@order_by | \%order_by )
3669
3670 =back
3671
3672 Which column(s) to order the results by.
3673
3674 [The full list of suitable values is documented in
3675 L<SQL::Abstract/"ORDER BY CLAUSES">; the following is a summary of
3676 common options.]
3677
3678 If a single column name, or an arrayref of names is supplied, the
3679 argument is passed through directly to SQL. The hashref syntax allows
3680 for connection-agnostic specification of ordering direction:
3681
3682  For descending order:
3683
3684   order_by => { -desc => [qw/col1 col2 col3/] }
3685
3686  For explicit ascending order:
3687
3688   order_by => { -asc => 'col' }
3689
3690 The old scalarref syntax (i.e. order_by => \'year DESC') is still
3691 supported, although you are strongly encouraged to use the hashref
3692 syntax as outlined above.
3693
3694 =head2 columns
3695
3696 =over 4
3697
3698 =item Value: \@columns
3699
3700 =back
3701
3702 Shortcut to request a particular set of columns to be retrieved. Each
3703 column spec may be a string (a table column name), or a hash (in which
3704 case the key is the C<as> value, and the value is used as the C<select>
3705 expression). Adds C<me.> onto the start of any column without a C<.> in
3706 it and sets C<select> from that, then auto-populates C<as> from
3707 C<select> as normal. (You may also use the C<cols> attribute, as in
3708 earlier versions of DBIC.)
3709
3710 Essentially C<columns> does the same as L</select> and L</as>.
3711
3712     columns => [ 'foo', { bar => 'baz' } ]
3713
3714 is the same as
3715
3716     select => [qw/foo baz/],
3717     as => [qw/foo bar/]
3718
3719 =head2 +columns
3720
3721 =over 4
3722
3723 =item Value: \@columns
3724
3725 =back
3726
3727 Indicates additional columns to be selected from storage. Works the same
3728 as L</columns> but adds columns to the selection. (You may also use the
3729 C<include_columns> attribute, as in earlier versions of DBIC). For
3730 example:-
3731
3732   $schema->resultset('CD')->search(undef, {
3733     '+columns' => ['artist.name'],
3734     join => ['artist']
3735   });
3736
3737 would return all CDs and include a 'name' column to the information
3738 passed to object inflation. Note that the 'artist' is the name of the
3739 column (or relationship) accessor, and 'name' is the name of the column
3740 accessor in the related table.
3741
3742 =head2 include_columns
3743
3744 =over 4
3745
3746 =item Value: \@columns
3747
3748 =back
3749
3750 Deprecated.  Acts as a synonym for L</+columns> for backward compatibility.
3751
3752 =head2 select
3753
3754 =over 4
3755
3756 =item Value: \@select_columns
3757
3758 =back
3759
3760 Indicates which columns should be selected from the storage. You can use
3761 column names, or in the case of RDBMS back ends, function or stored procedure
3762 names:
3763
3764   $rs = $schema->resultset('Employee')->search(undef, {
3765     select => [
3766       'name',
3767       { count => 'employeeid' },
3768       { max => { length => 'name' }, -as => 'longest_name' }
3769     ]
3770   });
3771
3772   # Equivalent SQL
3773   SELECT name, COUNT( employeeid ), MAX( LENGTH( name ) ) AS longest_name FROM employee
3774
3775 B<NOTE:> You will almost always need a corresponding L</as> attribute when you
3776 use L</select>, to instruct DBIx::Class how to store the result of the column.
3777 Also note that the L</as> attribute has nothing to do with the SQL-side 'AS'
3778 identifier aliasing. You can however alias a function, so you can use it in
3779 e.g. an C<ORDER BY> clause. This is done via the C<-as> B<select function
3780 attribute> supplied as shown in the example above.
3781
3782 =head2 +select
3783
3784 =over 4
3785
3786 Indicates additional columns to be selected from storage.  Works the same as
3787 L</select> but adds columns to the default selection, instead of specifying
3788 an explicit list.
3789
3790 =back
3791
3792 =head2 +as
3793
3794 =over 4
3795
3796 Indicates additional column names for those added via L</+select>. See L</as>.
3797
3798 =back
3799
3800 =head2 as
3801
3802 =over 4
3803
3804 =item Value: \@inflation_names
3805
3806 =back
3807
3808 Indicates column names for object inflation. That is L</as> indicates the
3809 slot name in which the column value will be stored within the
3810 L<Row|DBIx::Class::Row> object. The value will then be accessible via this
3811 identifier by the C<get_column> method (or via the object accessor B<if one
3812 with the same name already exists>) as shown below. The L</as> attribute has
3813 B<nothing to do> with the SQL-side C<AS>. See L</select> for details.
3814
3815   $rs = $schema->resultset('Employee')->search(undef, {
3816     select => [
3817       'name',
3818       { count => 'employeeid' },
3819       { max => { length => 'name' }, -as => 'longest_name' }
3820     ],
3821     as => [qw/
3822       name
3823       employee_count
3824       max_name_length
3825     /],
3826   });
3827
3828 If the object against which the search is performed already has an accessor
3829 matching a column name specified in C<as>, the value can be retrieved using
3830 the accessor as normal:
3831
3832   my $name = $employee->name();
3833
3834 If on the other hand an accessor does not exist in the object, you need to
3835 use C<get_column> instead:
3836
3837   my $employee_count = $employee->get_column('employee_count');
3838
3839 You can create your own accessors if required - see
3840 L<DBIx::Class::Manual::Cookbook> for details.
3841
3842 =head2 join
3843
3844 =over 4
3845
3846 =item Value: ($rel_name | \@rel_names | \%rel_names)
3847
3848 =back
3849
3850 Contains a list of relationships that should be joined for this query.  For
3851 example:
3852
3853   # Get CDs by Nine Inch Nails
3854   my $rs = $schema->resultset('CD')->search(
3855     { 'artist.name' => 'Nine Inch Nails' },
3856     { join => 'artist' }
3857   );
3858
3859 Can also contain a hash reference to refer to the other relation's relations.
3860 For example:
3861
3862   package MyApp::Schema::Track;
3863   use base qw/DBIx::Class/;
3864   __PACKAGE__->table('track');
3865   __PACKAGE__->add_columns(qw/trackid cd position title/);
3866   __PACKAGE__->set_primary_key('trackid');
3867   __PACKAGE__->belongs_to(cd => 'MyApp::Schema::CD');
3868   1;
3869
3870   # In your application
3871   my $rs = $schema->resultset('Artist')->search(
3872     { 'track.title' => 'Teardrop' },
3873     {
3874       join     => { cd => 'track' },
3875       order_by => 'artist.name',
3876     }
3877   );
3878
3879 You need to use the relationship (not the table) name in  conditions,
3880 because they are aliased as such. The current table is aliased as "me", so
3881 you need to use me.column_name in order to avoid ambiguity. For example:
3882
3883   # Get CDs from 1984 with a 'Foo' track
3884   my $rs = $schema->resultset('CD')->search(
3885     {
3886       'me.year' => 1984,
3887       'tracks.name' => 'Foo'
3888     },
3889     { join => 'tracks' }
3890   );
3891
3892 If the same join is supplied twice, it will be aliased to <rel>_2 (and
3893 similarly for a third time). For e.g.
3894
3895   my $rs = $schema->resultset('Artist')->search({
3896     'cds.title'   => 'Down to Earth',
3897     'cds_2.title' => 'Popular',
3898   }, {
3899     join => [ qw/cds cds/ ],
3900   });
3901
3902 will return a set of all artists that have both a cd with title 'Down
3903 to Earth' and a cd with title 'Popular'.
3904
3905 If you want to fetch related objects from other tables as well, see C<prefetch>
3906 below.
3907
3908 For more help on using joins with search, see L<DBIx::Class::Manual::Joining>.
3909
3910 =head2 prefetch
3911
3912 =over 4
3913
3914 =item Value: ($rel_name | \@rel_names | \%rel_names)
3915
3916 =back
3917
3918 Contains one or more relationships that should be fetched along with
3919 the main query (when they are accessed afterwards the data will
3920 already be available, without extra queries to the database).  This is
3921 useful for when you know you will need the related objects, because it
3922 saves at least one query:
3923
3924   my $rs = $schema->resultset('Tag')->search(
3925     undef,
3926     {
3927       prefetch => {
3928         cd => 'artist'
3929       }
3930     }
3931   );
3932
3933 The initial search results in SQL like the following:
3934
3935   SELECT tag.*, cd.*, artist.* FROM tag
3936   JOIN cd ON tag.cd = cd.cdid
3937   JOIN artist ON cd.artist = artist.artistid
3938
3939 L<DBIx::Class> has no need to go back to the database when we access the
3940 C<cd> or C<artist> relationships, which saves us two SQL statements in this
3941 case.
3942
3943 Simple prefetches will be joined automatically, so there is no need
3944 for a C<join> attribute in the above search.
3945
3946 C<prefetch> can be used with the following relationship types: C<belongs_to>,
3947 C<has_one> (or if you're using C<add_relationship>, any relationship declared
3948 with an accessor type of 'single' or 'filter'). A more complex example that
3949 prefetches an artists cds, the tracks on those cds, and the tags associated
3950 with that artist is given below (assuming many-to-many from artists to tags):
3951
3952  my $rs = $schema->resultset('Artist')->search(
3953    undef,
3954    {
3955      prefetch => [
3956        { cds => 'tracks' },
3957        { artist_tags => 'tags' }
3958      ]
3959    }
3960  );
3961
3962
3963 B<NOTE:> If you specify a C<prefetch> attribute, the C<join> and C<select>
3964 attributes will be ignored.
3965
3966 B<CAVEATs>: Prefetch does a lot of deep magic. As such, it may not behave
3967 exactly as you might expect.
3968
3969 =over 4
3970
3971 =item *
3972
3973 Prefetch uses the L</cache> to populate the prefetched relationships. This
3974 may or may not be what you want.
3975
3976 =item *
3977
3978 If you specify a condition on a prefetched relationship, ONLY those
3979 rows that match the prefetched condition will be fetched into that relationship.
3980 This means that adding prefetch to a search() B<may alter> what is returned by
3981 traversing a relationship. So, if you have C<< Artist->has_many(CDs) >> and you do
3982
3983   my $artist_rs = $schema->resultset('Artist')->search({
3984       'cds.year' => 2008,
3985   }, {
3986       join => 'cds',
3987   });
3988
3989   my $count = $artist_rs->first->cds->count;
3990
3991   my $artist_rs_prefetch = $artist_rs->search( {}, { prefetch => 'cds' } );
3992
3993   my $prefetch_count = $artist_rs_prefetch->first->cds->count;
3994
3995   cmp_ok( $count, '==', $prefetch_count, "Counts should be the same" );
3996
3997 that cmp_ok() may or may not pass depending on the datasets involved. This
3998 behavior may or may not survive the 0.09 transition.
3999
4000 =back
4001
4002 =head2 page
4003
4004 =over 4
4005
4006 =item Value: $page
4007
4008 =back
4009
4010 Makes the resultset paged and specifies the page to retrieve. Effectively
4011 identical to creating a non-pages resultset and then calling ->page($page)
4012 on it.
4013
4014 If L<rows> attribute is not specified it defaults to 10 rows per page.
4015
4016 When you have a paged resultset, L</count> will only return the number
4017 of rows in the page. To get the total, use the L</pager> and call
4018 C<total_entries> on it.
4019
4020 =head2 rows
4021
4022 =over 4
4023
4024 =item Value: $rows
4025
4026 =back
4027
4028 Specifies the maximum number of rows for direct retrieval or the number of
4029 rows per page if the page attribute or method is used.
4030
4031 =head2 offset
4032
4033 =over 4
4034
4035 =item Value: $offset
4036
4037 =back
4038
4039 Specifies the (zero-based) row number for the  first row to be returned, or the
4040 of the first row of the first page if paging is used.
4041
4042 =head2 group_by
4043
4044 =over 4
4045
4046 =item Value: \@columns
4047
4048 =back
4049
4050 A arrayref of columns to group by. Can include columns of joined tables.
4051
4052   group_by => [qw/ column1 column2 ... /]
4053
4054 =head2 having
4055
4056 =over 4
4057
4058 =item Value: $condition
4059
4060 =back
4061
4062 HAVING is a select statement attribute that is applied between GROUP BY and
4063 ORDER BY. It is applied to the after the grouping calculations have been
4064 done.
4065
4066   having => { 'count_employee' => { '>=', 100 } }
4067
4068 or with an in-place function in which case literal SQL is required:
4069
4070   having => \[ 'count(employee) >= ?', [ count => 100 ] ]
4071
4072 =head2 distinct
4073
4074 =over 4
4075
4076 =item Value: (0 | 1)
4077
4078 =back
4079
4080 Set to 1 to group by all columns. If the resultset already has a group_by
4081 attribute, this setting is ignored and an appropriate warning is issued.
4082
4083 =head2 where
4084
4085 =over 4
4086
4087 Adds to the WHERE clause.
4088
4089   # only return rows WHERE deleted IS NULL for all searches
4090   __PACKAGE__->resultset_attributes({ where => { deleted => undef } }); )
4091
4092 Can be overridden by passing C<< { where => undef } >> as an attribute
4093 to a resultset.
4094
4095 =back
4096
4097 =head2 cache
4098
4099 Set to 1 to cache search results. This prevents extra SQL queries if you
4100 revisit rows in your ResultSet:
4101
4102   my $resultset = $schema->resultset('Artist')->search( undef, { cache => 1 } );
4103
4104   while( my $artist = $resultset->next ) {
4105     ... do stuff ...
4106   }
4107
4108   $rs->first; # without cache, this would issue a query
4109
4110 By default, searches are not cached.
4111
4112 For more examples of using these attributes, see
4113 L<DBIx::Class::Manual::Cookbook>.
4114
4115 =head2 for
4116
4117 =over 4
4118
4119 =item Value: ( 'update' | 'shared' )
4120
4121 =back
4122
4123 Set to 'update' for a SELECT ... FOR UPDATE or 'shared' for a SELECT
4124 ... FOR SHARED.
4125
4126 =cut
4127
4128 1;