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